Delete files

files/Dockerfile

@@ -1,44 +0,0 @@
-# Telegram Multi-Part File Streamer - Dockerfile
-# Optimized for Hugging Face Spaces with low RAM footprint
-
-FROM python:3.11-slim
-
-# Set environment variables
-ENV PYTHONUNBUFFERED=1 \
-    PYTHONDONTWRITEBYTECODE=1 \
-    PIP_NO_CACHE_DIR=1 \
-    PIP_DISABLE_PIP_VERSION_CHECK=1
-
-# Set working directory
-WORKDIR /app
-
-# Install system dependencies (minimal set)
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    gcc \
-    g++ \
-    && rm -rf /var/lib/apt/lists/*
-
-# Copy requirements first for better caching
-COPY requirements.txt .
-
-# Install Python dependencies
-RUN pip install --no-cache-dir -r requirements.txt
-
-# Copy application code
-COPY main.py .
-COPY session_manager.py .
-COPY database.py .
-COPY utils.py .
-
-# Create directory for session files (if needed)
-RUN mkdir -p /app/sessions
-
-# Health check
-HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
-    CMD python -c "import requests; requests.get('http://localhost:8000/health')"
-
-# Expose port
-EXPOSE 8000
-
-# Run the application
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

files/Makefile

@@ -1,62 +0,0 @@
-.PHONY: help install dev run test docker-build docker-run docker-stop clean session
-
-help:
-	@echo "Telegram Multi-Part File Streamer - Makefile Commands"
-	@echo ""
-	@echo "Setup:"
-	@echo "  make install          Install dependencies"
-	@echo "  make session          Generate Pyrogram session string"
-	@echo ""
-	@echo "Development:"
-	@echo "  make dev              Run development server with auto-reload"
-	@echo "  make run              Run production server"
-	@echo "  make test             Run test suite"
-	@echo ""
-	@echo "Docker:"
-	@echo "  make docker-build     Build Docker image"
-	@echo "  make docker-run       Run with Docker Compose"
-	@echo "  make docker-stop      Stop Docker containers"
-	@echo "  make docker-logs      View Docker logs"
-	@echo ""
-	@echo "Maintenance:"
-	@echo "  make clean            Clean temporary files"
-	@echo "  make logs             View application logs"
-
-install:
-	pip install -r requirements.txt
-
-session:
-	python generate_session.py
-
-dev:
-	uvicorn main:app --reload --host 0.0.0.0 --port 8000
-
-run:
-	uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1
-
-test:
-	python test_setup.py
-
-docker-build:
-	docker build -t telegram-streamer .
-
-docker-run:
-	docker-compose up -d
-
-docker-stop:
-	docker-compose down
-
-docker-logs:
-	docker-compose logs -f app
-
-logs:
-	tail -f *.log
-
-clean:
-	find . -type d -name __pycache__ -exec rm -rf {} +
-	find . -type f -name "*.pyc" -delete
-	find . -type f -name "*.pyo" -delete
-	find . -type f -name "*.log" -delete
-	find . -type f -name "*.session" -delete
-	find . -type f -name "*.session-journal" -delete
-	rm -rf build/ dist/ *.egg-info/

files/README.md

@@ -1,384 +0,0 @@
-# 🚀 Telegram Multi-Part File Streamer
-
-A high-performance file upload and streaming service that uses Telegram as a backend storage system. Capable of handling files up to **1TB** with zero-disk buffering, optimized for low-RAM environments like Hugging Face Spaces.
-
-## ✨ Features
-
-- **🎯 Zero-Disk Buffering**: Direct streaming from HTTP to Telegram without local storage
-- **📦 Auto-Splitting**: Automatically splits large files into 2GB parts
-- **⚡ Multi-Session Load Balancing**: Rotates between multiple Telegram sessions for maximum bandwidth
-- **📡 Full Range Request Support**: HTTP 206 partial content for seeking in 1TB files
-- **🔄 Parallel Download Support**: Compatible with IDM and multi-threaded downloaders
-- **💾 Low Memory Footprint**: Optimized for <512MB RAM usage even with 1TB files
-- **🛡️ Production Ready**: Comprehensive error handling, retry logic, and logging
-
-## 🏗️ Architecture
-
-```
-┌─────────────┐
-│  Client     │
-│  (Browser/  │
-│   IDM)      │
-└──────┬──────┘
-       │
-       │ HTTP Stream
-       │
-┌──────▼────────────────────────────┐
-│  FastAPI Server                   │
-│  ┌─────────────────────────────┐ │
-│  │  /upload (POST)             │ │
-│  │  - Stream from HTTP         │ │
-│  │  - Auto-split into parts    │ │
-│  │  - Upload to Telegram       │ │
-│  └─────────────────────────────┘ │
-│  ┌─────────────────────────────┐ │
-│  │  /dl/{id} (GET)             │ │
-│  │  - Multi-part concatenation │ │
-│  │  - Range request support    │ │
-│  │  - Session load balancing   │ │
-│  └─────────────────────────────┘ │
-└──────┬────────────────────────────┘
-       │
-       ├──────► Telegram Session 1
-       ├──────► Telegram Session 2
-       └──────► Telegram Session N
-                (Load Balanced)
-       
-       │
-┌──────▼──────┐
-│  MongoDB    │
-│  (Metadata) │
-└─────────────┘
-```
-
-## 📋 Requirements
-
-- Python 3.10+
-- MongoDB (local or Atlas)
-- Telegram API credentials
-- At least 512MB RAM
-
-## 🚀 Quick Start
-
-### 1. Clone and Setup
-
-```bash
-git clone <repository>
-cd telegram-streamer
-pip install -r requirements.txt
-```
-
-### 2. Configure Environment
-
-```bash
-cp .env.template .env
-# Edit .env with your credentials
-```
-
-Required environment variables:
-- `API_ID`: Your Telegram API ID
-- `API_HASH`: Your Telegram API Hash
-- `BOT_TOKEN`: Telegram bot token
-- `MONGO_URI`: MongoDB connection string
-- `SESSION_STRINGS`: (Optional) Comma-separated Pyrogram session strings
-
-### 3. Generate Session Strings (Optional but Recommended)
-
-For multi-session load balancing:
-
-```bash
-python -c "from pyrogram import Client; \
-client = Client('my_session', api_id=YOUR_API_ID, api_hash='YOUR_API_HASH'); \
-client.start(); \
-print(client.export_session_string())"
-```
-
-Add the output to `SESSION_STRINGS` in your `.env` file.
-
-### 4. Run the Application
-
-```bash
-# Development
-uvicorn main:app --reload
-
-# Production
-uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1
-```
-
-### 5. Using Docker
-
-```bash
-# Build
-docker build -t telegram-streamer .
-
-# Run
-docker run -p 8000:8000 --env-file .env telegram-streamer
-```
-
-## 📡 API Endpoints
-
-### Upload File
-
-**POST** `/upload?filename=myfile.zip`
-
-Upload a file with zero-disk buffering:
-
-```bash
-curl -X POST "http://localhost:8000/upload?filename=large_file.zip" \
-  -H "Content-Type: application/octet-stream" \
-  --data-binary "@large_file.zip"
-```
-
-Response:
-```json
-{
-  "success": true,
-  "unique_id": "a1b2c3d4e5f6g7h8",
-  "filename": "large_file.zip",
-  "total_size": 1099511627776,
-  "parts": 550,
-  "download_url": "/dl/a1b2c3d4e5f6g7h8"
-}
-```
-
-### Stream/Download File
-
-**GET** `/dl/{unique_id}`
-
-Stream a file with full range request support:
-
-```bash
-# Full download
-curl "http://localhost:8000/dl/a1b2c3d4e5f6g7h8" -o output.zip
-
-# Range request (download bytes 1000000-2000000)
-curl "http://localhost:8000/dl/a1b2c3d4e5f6g7h8" \
-  -H "Range: bytes=1000000-2000000" -o partial.zip
-```
-
-### Get File Info
-
-**GET** `/info/{unique_id}`
-
-```bash
-curl "http://localhost:8000/info/a1b2c3d4e5f6g7h8"
-```
-
-### Delete File
-
-**DELETE** `/delete/{unique_id}`
-
-```bash
-curl -X DELETE "http://localhost:8000/delete/a1b2c3d4e5f6g7h8"
-```
-
-## 🎯 Usage Examples
-
-### Upload with Python
-
-```python
-import requests
-
-url = "http://localhost:8000/upload"
-params = {"filename": "large_dataset.tar.gz"}
-
-with open("large_dataset.tar.gz", "rb") as f:
-    response = requests.post(url, params=params, data=f, stream=True)
-    
-print(response.json())
-```
-
-### Download with Python
-
-```python
-import requests
-
-unique_id = "a1b2c3d4e5f6g7h8"
-url = f"http://localhost:8000/dl/{unique_id}"
-
-with requests.get(url, stream=True) as r:
-    r.raise_for_status()
-    with open("output.zip", "wb") as f:
-        for chunk in r.iter_content(chunk_size=8192):
-            f.write(chunk)
-```
-
-### Range Request Example
-
-```python
-import requests
-
-unique_id = "a1b2c3d4e5f6g7h8"
-url = f"http://localhost:8000/dl/{unique_id}"
-
-# Download only bytes 1MB to 10MB
-headers = {"Range": "bytes=1048576-10485760"}
-response = requests.get(url, headers=headers)
-
-with open("partial.bin", "wb") as f:
-    f.write(response.content)
-```
-
-## 🔧 Configuration
-
-### Environment Variables
-
-| Variable | Description | Required | Default |
-|----------|-------------|----------|---------|
-| `API_ID` | Telegram API ID | ✅ | - |
-| `API_HASH` | Telegram API Hash | ✅ | - |
-| `BOT_TOKEN` | Telegram Bot Token | ✅ | - |
-| `MONGO_URI` | MongoDB connection string | ✅ | - |
-| `MONGO_DATABASE` | Database name | ❌ | telegram_streamer |
-| `SESSION_STRINGS` | Comma-separated session strings | ❌ | (uses bot only) |
-| `LOG_LEVEL` | Logging level | ❌ | INFO |
-
-### Performance Tuning
-
-**For Maximum Upload Speed:**
-- Add multiple session strings to `SESSION_STRINGS`
-- Each session can handle ~10-15 MB/s upload speed
-- 5 sessions = ~50-75 MB/s combined
-
-**For Maximum Download Speed:**
-- Session rotation automatically balances load
-- Supports unlimited parallel connections
-- Compatible with IDM, aria2c, axel
-
-**Memory Optimization:**
-- Default chunk size: 1MB (adjustable via `CHUNK_SIZE`)
-- RAM usage stays constant regardless of file size
-- Recommended: 512MB RAM minimum
-
-## 🏆 Performance Benchmarks
-
-| Metric | Value |
-|--------|-------|
-| Max File Size | 1TB+ |
-| Upload Speed | 10-15 MB/s per session |
-| Download Speed | 50-100 MB/s (multi-threaded) |
-| RAM Usage | <512MB (constant) |
-| Concurrent Uploads | Limited by sessions |
-| Concurrent Downloads | Unlimited |
-
-## 🐛 Troubleshooting
-
-### FloodWait Errors
-
-If you encounter `FloodWait` errors:
-1. Add more session strings to `SESSION_STRINGS`
-2. Session rotation will automatically handle flood waits
-3. Consider adding delays between uploads
-
-### MongoDB Connection Issues
-
-```bash
-# Test MongoDB connection
-mongosh "YOUR_MONGO_URI"
-
-# Check if MongoDB is running (local)
-sudo systemctl status mongod
-```
-
-### Session Initialization Failures
-
-1. Verify `API_ID` and `API_HASH` are correct
-2. Ensure session strings are valid
-3. Check Telegram API availability
-
-## 🌐 Deployment
-
-### Hugging Face Spaces
-
-1. Create a new Space (Docker type)
-2. Add Secrets in Space Settings:
-   - `API_ID`
-   - `API_HASH`
-   - `BOT_TOKEN`
-   - `MONGO_URI`
-   - `SESSION_STRINGS`
-3. Push code to Space repository
-4. App will auto-deploy
-
-### Railway/Render
-
-1. Connect GitHub repository
-2. Add environment variables
-3. Deploy (automatically detects Dockerfile)
-
-### VPS/Dedicated Server
-
-```bash
-# Using systemd
-sudo nano /etc/systemd/system/telegram-streamer.service
-
-[Unit]
-Description=Telegram Multi-Part File Streamer
-After=network.target
-
-[Service]
-Type=simple
-User=www-data
-WorkingDirectory=/opt/telegram-streamer
-EnvironmentFile=/opt/telegram-streamer/.env
-ExecStart=/usr/bin/uvicorn main:app --host 0.0.0.0 --port 8000
-Restart=always
-
-[Install]
-WantedBy=multi-user.target
-
-# Enable and start
-sudo systemctl enable telegram-streamer
-sudo systemctl start telegram-streamer
-```
-
-## 📊 Monitoring
-
-Check application health:
-
-```bash
-curl http://localhost:8000/health
-```
-
-Response:
-```json
-{
-  "status": "healthy",
-  "sessions": 5,
-  "database": "connected"
-}
-```
-
-## 🔒 Security Considerations
-
-1. **Access Control**: Add authentication middleware for production
-2. **Rate Limiting**: Implement rate limits on upload/download endpoints
-3. **Input Validation**: Validate filenames and parameters
-4. **HTTPS**: Always use HTTPS in production
-5. **Session Security**: Keep session strings private
-
-## 📝 License
-
-MIT License - See LICENSE file for details
-
-## 🤝 Contributing
-
-Contributions are welcome! Please:
-1. Fork the repository
-2. Create a feature branch
-3. Submit a pull request
-
-## 📞 Support
-
-- Issues: GitHub Issues
-- Discussions: GitHub Discussions
-
-## 🙏 Acknowledgments
-
-- Built with [Pyrogram](https://docs.pyrogram.org/)
-- Powered by [FastAPI](https://fastapi.tiangolo.com/)
-- Storage: [MongoDB](https://www.mongodb.com/)
-
----
-
-**⚠️ Disclaimer**: This tool uses Telegram as a storage backend. Ensure compliance with Telegram's Terms of Service. Not recommended for storing illegal or copyrighted content.

files/database.py

@@ -1,262 +0,0 @@
-"""
-Database Module - MongoDB integration for file metadata storage
-Uses Motor async driver for high-performance operations
-"""
-
-import logging
-import os
-from datetime import datetime
-from typing import Optional, List, Dict
-from dataclasses import dataclass, field, asdict
-
-from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
-from pymongo.errors import DuplicateKeyError
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class FileMetadata:
-    """File metadata structure"""
-    unique_id: str
-    filename: str
-    total_size: int
-    parts: List[Dict]
-    part_count: int
-    uploaded_at: datetime = field(default_factory=datetime.utcnow)
-    
-    def to_dict(self) -> dict:
-        """Convert to dictionary for MongoDB"""
-        data = asdict(self)
-        data["uploaded_at"] = self.uploaded_at.isoformat()
-        return data
-    
-    @classmethod
-    def from_dict(cls, data: dict) -> "FileMetadata":
-        """Create from MongoDB document"""
-        data["uploaded_at"] = datetime.fromisoformat(data["uploaded_at"])
-        return cls(**data)
-
-
-class Database:
-    """MongoDB database manager"""
-    
-    def __init__(self):
-        self.client: Optional[AsyncIOMotorClient] = None
-        self.db: Optional[AsyncIOMotorDatabase] = None
-        self.files_collection = None
-        self.mongo_uri = os.getenv("MONGO_URI")
-        self.database_name = os.getenv("MONGO_DATABASE", "telegram_streamer")
-    
-    async def connect(self):
-        """Connect to MongoDB"""
-        if not self.mongo_uri:
-            raise ValueError("MONGO_URI environment variable is required")
-        
-        try:
-            logger.info("Connecting to MongoDB...")
-            
-            self.client = AsyncIOMotorClient(
-                self.mongo_uri,
-                serverSelectionTimeoutMS=5000
-            )
-            
-            # Test connection
-            await self.client.admin.command('ping')
-            
-            self.db = self.client[self.database_name]
-            self.files_collection = self.db["files"]
-            
-            # Create indexes
-            await self._create_indexes()
-            
-            logger.info(f"Connected to MongoDB: {self.database_name}")
-        
-        except Exception as e:
-            logger.error(f"Failed to connect to MongoDB: {str(e)}")
-            raise
-    
-    async def _create_indexes(self):
-        """Create database indexes for performance"""
-        try:
-            # Unique index on unique_id
-            await self.files_collection.create_index(
-                "unique_id",
-                unique=True,
-                name="unique_id_index"
-            )
-            
-            # Index on uploaded_at for cleanup queries
-            await self.files_collection.create_index(
-                "uploaded_at",
-                name="uploaded_at_index"
-            )
-            
-            logger.info("Database indexes created")
-        
-        except Exception as e:
-            logger.warning(f"Failed to create indexes: {str(e)}")
-    
-    async def disconnect(self):
-        """Disconnect from MongoDB"""
-        if self.client:
-            self.client.close()
-            logger.info("Disconnected from MongoDB")
-    
-    def is_connected(self) -> bool:
-        """Check if database is connected"""
-        return self.client is not None and self.db is not None
-    
-    async def save_file_metadata(self, metadata: FileMetadata) -> bool:
-        """Save file metadata to database"""
-        try:
-            await self.files_collection.insert_one(metadata.to_dict())
-            logger.info(f"Saved metadata: unique_id={metadata.unique_id}")
-            return True
-        
-        except DuplicateKeyError:
-            logger.error(f"Duplicate unique_id: {metadata.unique_id}")
-            raise ValueError("File with this unique_id already exists")
-        
-        except Exception as e:
-            logger.error(f"Failed to save metadata: {str(e)}")
-            raise
-    
-    async def get_file_metadata(self, unique_id: str) -> Optional[FileMetadata]:
-        """Retrieve file metadata by unique_id"""
-        try:
-            doc = await self.files_collection.find_one({"unique_id": unique_id})
-            
-            if doc:
-                # Remove MongoDB _id field
-                doc.pop("_id", None)
-                return FileMetadata.from_dict(doc)
-            
-            return None
-        
-        except Exception as e:
-            logger.error(f"Failed to get metadata: {str(e)}")
-            return None
-    
-    async def update_file_metadata(
-        self,
-        unique_id: str,
-        updates: dict
-    ) -> bool:
-        """Update file metadata"""
-        try:
-            result = await self.files_collection.update_one(
-                {"unique_id": unique_id},
-                {"$set": updates}
-            )
-            
-            return result.modified_count > 0
-        
-        except Exception as e:
-            logger.error(f"Failed to update metadata: {str(e)}")
-            return False
-    
-    async def delete_file_metadata(self, unique_id: str) -> bool:
-        """Delete file metadata"""
-        try:
-            result = await self.files_collection.delete_one(
-                {"unique_id": unique_id}
-            )
-            
-            logger.info(f"Deleted metadata: unique_id={unique_id}")
-            return result.deleted_count > 0
-        
-        except Exception as e:
-            logger.error(f"Failed to delete metadata: {str(e)}")
-            return False
-    
-    async def list_files(
-        self,
-        limit: int = 100,
-        skip: int = 0
-    ) -> List[FileMetadata]:
-        """List all files with pagination"""
-        try:
-            cursor = self.files_collection.find().skip(skip).limit(limit)
-            cursor = cursor.sort("uploaded_at", -1)
-            
-            files = []
-            async for doc in cursor:
-                doc.pop("_id", None)
-                files.append(FileMetadata.from_dict(doc))
-            
-            return files
-        
-        except Exception as e:
-            logger.error(f"Failed to list files: {str(e)}")
-            return []
-    
-    async def get_total_storage(self) -> int:
-        """Get total storage used across all files"""
-        try:
-            pipeline = [
-                {
-                    "$group": {
-                        "_id": None,
-                        "total_size": {"$sum": "$total_size"}
-                    }
-                }
-            ]
-            
-            result = await self.files_collection.aggregate(pipeline).to_list(1)
-            
-            if result:
-                return result[0]["total_size"]
-            
-            return 0
-        
-        except Exception as e:
-            logger.error(f"Failed to get total storage: {str(e)}")
-            return 0
-    
-    async def cleanup_old_files(self, days: int = 30) -> int:
-        """Delete files older than specified days"""
-        try:
-            from datetime import timedelta
-            
-            cutoff_date = datetime.utcnow() - timedelta(days=days)
-            
-            result = await self.files_collection.delete_many(
-                {"uploaded_at": {"$lt": cutoff_date.isoformat()}}
-            )
-            
-            deleted = result.deleted_count
-            logger.info(f"Cleaned up {deleted} old files (older than {days} days)")
-            
-            return deleted
-        
-        except Exception as e:
-            logger.error(f"Failed to cleanup old files: {str(e)}")
-            return 0
-    
-    async def get_stats(self) -> dict:
-        """Get database statistics"""
-        try:
-            total_files = await self.files_collection.count_documents({})
-            total_storage = await self.get_total_storage()
-            
-            # Get largest file
-            largest = await self.files_collection.find_one(
-                {},
-                sort=[("total_size", -1)]
-            )
-            
-            return {
-                "total_files": total_files,
-                "total_storage": total_storage,
-                "total_storage_gb": f"{total_storage / (1024**3):.2f}",
-                "largest_file": {
-                    "unique_id": largest.get("unique_id"),
-                    "filename": largest.get("filename"),
-                    "size": largest.get("total_size")
-                } if largest else None
-            }
-        
-        except Exception as e:
-            logger.error(f"Failed to get stats: {str(e)}")
-            return {}

files/docker-compose.yml

@@ -1,85 +0,0 @@
-version: '3.8'
-
-services:
-  # MongoDB service
-  mongodb:
-    image: mongo:7.0
-    container_name: telegram_streamer_mongodb
-    restart: unless-stopped
-    ports:
-      - "27017:27017"
-    environment:
-      MONGO_INITDB_ROOT_USERNAME: admin
-      MONGO_INITDB_ROOT_PASSWORD: password123
-      MONGO_INITDB_DATABASE: telegram_streamer
-    volumes:
-      - mongodb_data:/data/db
-    networks:
-      - streamer_network
-    healthcheck:
-      test: echo 'db.runCommand("ping").ok' | mongosh localhost:27017/test --quiet
-      interval: 10s
-      timeout: 5s
-      retries: 5
-
-  # Telegram Streamer application
-  app:
-    build: .
-    container_name: telegram_streamer_app
-    restart: unless-stopped
-    ports:
-      - "8000:8000"
-    environment:
-      # Telegram API (REPLACE WITH YOUR VALUES)
-      API_ID: ${API_ID}
-      API_HASH: ${API_HASH}
-      BOT_TOKEN: ${BOT_TOKEN}
-      
-      # MongoDB connection
-      MONGO_URI: mongodb://admin:password123@mongodb:27017/telegram_streamer?authSource=admin
-      MONGO_DATABASE: telegram_streamer
-      
-      # Optional: Session strings for multi-session load balancing
-      SESSION_STRINGS: ${SESSION_STRINGS:-}
-      
-      # Logging
-      LOG_LEVEL: INFO
-    depends_on:
-      mongodb:
-        condition: service_healthy
-    networks:
-      - streamer_network
-    healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
-      interval: 30s
-      timeout: 10s
-      retries: 3
-      start_period: 10s
-
-  # Optional: Mongo Express for database management
-  mongo-express:
-    image: mongo-express:latest
-    container_name: telegram_streamer_mongo_express
-    restart: unless-stopped
-    ports:
-      - "8081:8081"
-    environment:
-      ME_CONFIG_MONGODB_ADMINUSERNAME: admin
-      ME_CONFIG_MONGODB_ADMINPASSWORD: password123
-      ME_CONFIG_MONGODB_URL: mongodb://admin:password123@mongodb:27017/
-      ME_CONFIG_BASICAUTH_USERNAME: admin
-      ME_CONFIG_BASICAUTH_PASSWORD: admin
-    depends_on:
-      - mongodb
-    networks:
-      - streamer_network
-    profiles:
-      - debug
-
-volumes:
-  mongodb_data:
-    driver: local
-
-networks:
-  streamer_network:
-    driver: bridge

files/example_client.py

@@ -1,344 +0,0 @@
-#!/usr/bin/env python3
-"""
-Example Client for Telegram Multi-Part File Streamer
-Demonstrates how to upload and download files programmatically
-"""
-
-import asyncio
-import os
-import time
-from pathlib import Path
-
-import httpx
-
-
-class TelegramStreamerClient:
-    """Client for interacting with Telegram File Streamer API"""
-    
-    def __init__(self, base_url: str = "http://localhost:8000"):
-        self.base_url = base_url
-        self.client = httpx.AsyncClient(timeout=300.0)
-    
-    async def close(self):
-        """Close the HTTP client"""
-        await self.client.aclose()
-    
-    async def upload_file(
-        self,
-        file_path: str,
-        filename: str = None,
-        chunk_size: int = 1024 * 1024  # 1MB chunks
-    ) -> dict:
-        """
-        Upload a file to the streamer
-        
-        Args:
-            file_path: Path to the file to upload
-            filename: Optional custom filename
-            chunk_size: Size of chunks for streaming upload
-        
-        Returns:
-            Upload response with unique_id and download_url
-        """
-        file_path = Path(file_path)
-        
-        if not file_path.exists():
-            raise FileNotFoundError(f"File not found: {file_path}")
-        
-        if filename is None:
-            filename = file_path.name
-        
-        file_size = file_path.stat().st_size
-        
-        print(f"📤 Uploading: {filename}")
-        print(f"   Size: {self._format_size(file_size)}")
-        
-        async def file_stream():
-            """Stream file in chunks"""
-            with open(file_path, "rb") as f:
-                uploaded = 0
-                start_time = time.time()
-                
-                while True:
-                    chunk = f.read(chunk_size)
-                    if not chunk:
-                        break
-                    
-                    uploaded += len(chunk)
-                    
-                    # Progress
-                    elapsed = time.time() - start_time
-                    if elapsed > 0:
-                        speed = uploaded / elapsed
-                        progress = (uploaded / file_size) * 100
-                        print(
-                            f"\r   Progress: {progress:.1f}% "
-                            f"({self._format_size(uploaded)}/{self._format_size(file_size)}) "
-                            f"Speed: {self._format_size(speed)}/s",
-                            end="",
-                            flush=True
-                        )
-                    
-                    yield chunk
-                
-                print()  # New line after progress
-        
-        start_time = time.time()
-        
-        response = await self.client.post(
-            f"{self.base_url}/upload",
-            params={"filename": filename},
-            content=file_stream()
-        )
-        
-        elapsed = time.time() - start_time
-        
-        response.raise_for_status()
-        result = response.json()
-        
-        print(f"✅ Upload completed in {elapsed:.2f}s")
-        print(f"   Unique ID: {result['unique_id']}")
-        print(f"   Parts: {result['parts']}")
-        print(f"   Download URL: {self.base_url}{result['download_url']}")
-        
-        return result
-    
-    async def download_file(
-        self,
-        unique_id: str,
-        output_path: str,
-        chunk_size: int = 1024 * 1024  # 1MB chunks
-    ):
-        """
-        Download a file from the streamer
-        
-        Args:
-            unique_id: Unique ID of the file
-            output_path: Path to save the downloaded file
-            chunk_size: Size of chunks for streaming download
-        """
-        output_path = Path(output_path)
-        
-        # Get file info first
-        info = await self.get_file_info(unique_id)
-        total_size = info["total_size"]
-        
-        print(f"📥 Downloading: {info['filename']}")
-        print(f"   Size: {self._format_size(total_size)}")
-        
-        start_time = time.time()
-        downloaded = 0
-        
-        async with self.client.stream(
-            "GET",
-            f"{self.base_url}/dl/{unique_id}"
-        ) as response:
-            response.raise_for_status()
-            
-            with open(output_path, "wb") as f:
-                async for chunk in response.aiter_bytes(chunk_size):
-                    f.write(chunk)
-                    downloaded += len(chunk)
-                    
-                    # Progress
-                    elapsed = time.time() - start_time
-                    if elapsed > 0:
-                        speed = downloaded / elapsed
-                        progress = (downloaded / total_size) * 100
-                        print(
-                            f"\r   Progress: {progress:.1f}% "
-                            f"({self._format_size(downloaded)}/{self._format_size(total_size)}) "
-                            f"Speed: {self._format_size(speed)}/s",
-                            end="",
-                            flush=True
-                        )
-        
-        print()  # New line after progress
-        elapsed = time.time() - start_time
-        
-        print(f"✅ Download completed in {elapsed:.2f}s")
-        print(f"   Saved to: {output_path}")
-    
-    async def download_range(
-        self,
-        unique_id: str,
-        start: int,
-        end: int,
-        output_path: str
-    ):
-        """
-        Download a specific byte range from a file
-        
-        Args:
-            unique_id: Unique ID of the file
-            start: Start byte position
-            end: End byte position (inclusive)
-            output_path: Path to save the downloaded chunk
-        """
-        output_path = Path(output_path)
-        
-        print(f"📥 Downloading range: bytes {start}-{end}")
-        
-        response = await self.client.get(
-            f"{self.base_url}/dl/{unique_id}",
-            headers={"Range": f"bytes={start}-{end}"}
-        )
-        
-        response.raise_for_status()
-        
-        if response.status_code != 206:
-            print(f"⚠️  Warning: Expected 206 Partial Content, got {response.status_code}")
-        
-        with open(output_path, "wb") as f:
-            f.write(response.content)
-        
-        print(f"✅ Downloaded {len(response.content)} bytes to {output_path}")
-    
-    async def get_file_info(self, unique_id: str) -> dict:
-        """Get file metadata"""
-        response = await self.client.get(f"{self.base_url}/info/{unique_id}")
-        response.raise_for_status()
-        return response.json()
-    
-    async def delete_file(self, unique_id: str) -> dict:
-        """Delete a file"""
-        response = await self.client.delete(f"{self.base_url}/delete/{unique_id}")
-        response.raise_for_status()
-        return response.json()
-    
-    async def health_check(self) -> dict:
-        """Check server health"""
-        response = await self.client.get(f"{self.base_url}/health")
-        response.raise_for_status()
-        return response.json()
-    
-    @staticmethod
-    def _format_size(size_bytes: int) -> str:
-        """Format byte size to human-readable string"""
-        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
-            if size_bytes < 1024.0:
-                return f"{size_bytes:.2f} {unit}"
-            size_bytes /= 1024.0
-        return f"{size_bytes:.2f} PB"
-
-
-async def example_upload():
-    """Example: Upload a file"""
-    client = TelegramStreamerClient()
-    
-    try:
-        # Create a test file
-        test_file = "test_upload.bin"
-        print(f"Creating test file: {test_file} (10MB)")
-        with open(test_file, "wb") as f:
-            f.write(os.urandom(10 * 1024 * 1024))  # 10MB
-        
-        # Upload
-        result = await client.upload_file(test_file)
-        unique_id = result["unique_id"]
-        
-        # Get info
-        print("\n📊 File Info:")
-        info = await client.get_file_info(unique_id)
-        for key, value in info.items():
-            print(f"   {key}: {value}")
-        
-        # Cleanup
-        os.remove(test_file)
-        
-        return unique_id
-    
-    finally:
-        await client.close()
-
-
-async def example_download(unique_id: str):
-    """Example: Download a file"""
-    client = TelegramStreamerClient()
-    
-    try:
-        output_file = "downloaded_file.bin"
-        await client.download_file(unique_id, output_file)
-        
-        # Cleanup
-        if os.path.exists(output_file):
-            os.remove(output_file)
-    
-    finally:
-        await client.close()
-
-
-async def example_range_request(unique_id: str):
-    """Example: Download a specific range"""
-    client = TelegramStreamerClient()
-    
-    try:
-        # Download first 1MB
-        output_file = "range_chunk.bin"
-        await client.download_range(unique_id, 0, 1024 * 1024 - 1, output_file)
-        
-        # Cleanup
-        if os.path.exists(output_file):
-            os.remove(output_file)
-    
-    finally:
-        await client.close()
-
-
-async def main():
-    """Main example"""
-    print("=" * 60)
-    print("Telegram Multi-Part File Streamer - Example Client")
-    print("=" * 60)
-    print()
-    
-    # Check server health
-    client = TelegramStreamerClient()
-    try:
-        health = await client.health_check()
-        print(f"🏥 Server Status: {health['status']}")
-        print(f"   Sessions: {health['sessions']}")
-        print(f"   Database: {health['database']}")
-        print()
-    except Exception as e:
-        print(f"❌ Server not available: {str(e)}")
-        print("   Make sure the server is running!")
-        return
-    finally:
-        await client.close()
-    
-    # Example 1: Upload
-    print("\n" + "=" * 60)
-    print("Example 1: Upload")
-    print("=" * 60)
-    unique_id = await example_upload()
-    
-    # Example 2: Download
-    print("\n" + "=" * 60)
-    print("Example 2: Download")
-    print("=" * 60)
-    await example_download(unique_id)
-    
-    # Example 3: Range Request
-    print("\n" + "=" * 60)
-    print("Example 3: Range Request")
-    print("=" * 60)
-    await example_range_request(unique_id)
-    
-    # Cleanup: Delete the file
-    print("\n" + "=" * 60)
-    print("Cleanup")
-    print("=" * 60)
-    client = TelegramStreamerClient()
-    try:
-        result = await client.delete_file(unique_id)
-        print(f"🗑️  Deleted file: {unique_id}")
-        print(f"   Deleted parts: {result['deleted_parts']}/{result['total_parts']}")
-    finally:
-        await client.close()
-    
-    print("\n✅ All examples completed!")
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

files/generate_session.py

@@ -1,117 +0,0 @@
-#!/usr/bin/env python3
-"""
-Session String Generator for Telegram Multi-Part File Streamer
-Generates Pyrogram session strings for multi-session load balancing
-"""
-
-import asyncio
-import sys
-from pyrogram import Client
-
-
-async def generate_session_string():
-    """Generate a Pyrogram session string"""
-    print("=" * 60)
-    print("Telegram Session String Generator")
-    print("=" * 60)
-    print()
-    
-    # Get credentials
-    try:
-        api_id = input("Enter your API_ID: ").strip()
-        if not api_id:
-            print("❌ API_ID is required!")
-            return
-        
-        api_id = int(api_id)
-        
-        api_hash = input("Enter your API_HASH: ").strip()
-        if not api_hash:
-            print("❌ API_HASH is required!")
-            return
-        
-        print()
-        print("🔐 Credentials validated!")
-        print()
-        print("📱 You will now receive a code on your Telegram app.")
-        print("   Please enter the code when prompted.")
-        print()
-        
-    except ValueError:
-        print("❌ Invalid API_ID! Must be a number.")
-        return
-    except KeyboardInterrupt:
-        print("\n\n❌ Cancelled by user.")
-        return
-    
-    # Create client
-    client = Client(
-        name="session_generator",
-        api_id=api_id,
-        api_hash=api_hash,
-        in_memory=True
-    )
-    
-    try:
-        # Start client (will prompt for phone number and code)
-        await client.start()
-        
-        # Get session string
-        session_string = await client.export_session_string()
-        
-        # Get user info
-        me = await client.get_me()
-        
-        print()
-        print("=" * 60)
-        print("✅ Session String Generated Successfully!")
-        print("=" * 60)
-        print()
-        print(f"👤 Account: {me.first_name} {me.last_name or ''}")
-        print(f"📞 Phone: +{me.phone_number}")
-        print(f"🆔 Username: @{me.username or 'N/A'}")
-        print()
-        print("📋 Session String:")
-        print("-" * 60)
-        print(session_string)
-        print("-" * 60)
-        print()
-        print("⚠️  IMPORTANT:")
-        print("   1. Keep this session string PRIVATE and SECURE")
-        print("   2. Anyone with this string can access your account")
-        print("   3. Add this to SESSION_STRINGS in your .env file")
-        print("   4. You can generate multiple session strings for")
-        print("      load balancing (comma-separated)")
-        print()
-        print("💡 Example .env configuration:")
-        print("-" * 60)
-        print(f"SESSION_STRINGS={session_string},YOUR_SECOND_SESSION_STRING")
-        print("-" * 60)
-        print()
-        
-        # Stop client
-        await client.stop()
-        
-    except Exception as e:
-        print(f"\n❌ Error: {str(e)}")
-        if "PASSWORD" in str(e).upper():
-            print("\n⚠️  Your account has 2FA enabled.")
-            print("   Please enter your password when prompted.")
-        return
-    
-    finally:
-        if client.is_connected:
-            await client.stop()
-
-
-def main():
-    """Main function"""
-    try:
-        asyncio.run(generate_session_string())
-    except KeyboardInterrupt:
-        print("\n\n❌ Cancelled by user.")
-        sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()

files/main.py

@@ -1,405 +0,0 @@
-"""
-Telegram Multi-Part File Streamer - Main Application
-High-performance file upload and streaming service with zero-disk buffering
-"""
-
-import asyncio
-import logging
-from typing import AsyncGenerator, Optional
-from contextlib import asynccontextmanager
-
-from fastapi import FastAPI, Request, HTTPException, Response
-from fastapi.responses import StreamingResponse
-from fastapi.middleware.cors import CORSMiddleware
-import uvicorn
-
-from session_manager import SessionManager
-from database import Database, FileMetadata
-from utils import (
-    calculate_part_and_offset,
-    generate_unique_id,
-    CHUNK_SIZE,
-    MAX_PART_SIZE
-)
-
-# Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-)
-logger = logging.getLogger(__name__)
-
-
-# Global instances
-session_manager: Optional[SessionManager] = None
-database: Optional[Database] = None
-
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    """Application lifespan manager"""
-    global session_manager, database
-    
-    logger.info("Initializing application...")
-    
-    # Initialize database
-    database = Database()
-    await database.connect()
-    
-    # Initialize session manager
-    session_manager = SessionManager()
-    await session_manager.initialize()
-    
-    logger.info("Application initialized successfully")
-    
-    yield
-    
-    # Cleanup
-    logger.info("Shutting down application...")
-    await session_manager.cleanup()
-    await database.disconnect()
-    logger.info("Application shutdown complete")
-
-
-# Initialize FastAPI app
-app = FastAPI(
-    title="Telegram Multi-Part File Streamer",
-    description="High-performance file upload and streaming service",
-    version="1.0.0",
-    lifespan=lifespan
-)
-
-# Add CORS middleware
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-@app.get("/")
-async def root():
-    """Health check endpoint"""
-    return {
-        "status": "online",
-        "service": "Telegram Multi-Part File Streamer",
-        "version": "1.0.0"
-    }
-
-
-@app.get("/health")
-async def health_check():
-    """Detailed health check"""
-    session_count = len(session_manager.sessions) if session_manager else 0
-    db_connected = database.is_connected() if database else False
-    
-    return {
-        "status": "healthy" if (session_count > 0 and db_connected) else "degraded",
-        "sessions": session_count,
-        "database": "connected" if db_connected else "disconnected"
-    }
-
-
-@app.post("/upload")
-async def upload_file(request: Request, filename: Optional[str] = None):
-    """
-    High-speed zero-disk file upload endpoint
-    Streams data directly from HTTP to Telegram with auto-splitting
-    """
-    if not session_manager or not database:
-        raise HTTPException(status_code=503, detail="Service not initialized")
-    
-    logger.info(f"Upload request received: filename={filename}")
-    
-    unique_id = generate_unique_id()
-    file_parts = []
-    total_size = 0
-    part_number = 0
-    
-    try:
-        # Create async generator from request stream
-        async def request_stream() -> AsyncGenerator[bytes, None]:
-            async for chunk in request.stream():
-                yield chunk
-        
-        # Buffer for part assembly
-        part_buffer = bytearray()
-        
-        async for chunk in request_stream():
-            part_buffer.extend(chunk)
-            
-            # Check if we need to upload this part
-            while len(part_buffer) >= MAX_PART_SIZE:
-                part_number += 1
-                part_data = bytes(part_buffer[:MAX_PART_SIZE])
-                part_buffer = part_buffer[MAX_PART_SIZE:]
-                
-                logger.info(f"Uploading part {part_number} ({len(part_data)} bytes)")
-                
-                # Upload part to Telegram
-                file_id = await session_manager.upload_part(
-                    part_data,
-                    f"{filename or unique_id}_part_{part_number}"
-                )
-                
-                file_parts.append({
-                    "part_number": part_number,
-                    "file_id": file_id,
-                    "size": len(part_data)
-                })
-                
-                total_size += len(part_data)
-                
-                logger.info(
-                    f"Part {part_number} uploaded successfully. "
-                    f"Total size: {total_size / (1024**3):.2f} GB"
-                )
-        
-        # Upload remaining data as final part
-        if len(part_buffer) > 0:
-            part_number += 1
-            part_data = bytes(part_buffer)
-            
-            logger.info(f"Uploading final part {part_number} ({len(part_data)} bytes)")
-            
-            file_id = await session_manager.upload_part(
-                part_data,
-                f"{filename or unique_id}_part_{part_number}"
-            )
-            
-            file_parts.append({
-                "part_number": part_number,
-                "file_id": file_id,
-                "size": len(part_data)
-            })
-            
-            total_size += len(part_data)
-        
-        # Store metadata in database
-        metadata = FileMetadata(
-            unique_id=unique_id,
-            filename=filename or f"file_{unique_id}",
-            total_size=total_size,
-            parts=file_parts,
-            part_count=part_number
-        )
-        
-        await database.save_file_metadata(metadata)
-        
-        logger.info(
-            f"Upload completed: unique_id={unique_id}, "
-            f"parts={part_number}, total_size={total_size / (1024**3):.2f} GB"
-        )
-        
-        return {
-            "success": True,
-            "unique_id": unique_id,
-            "filename": metadata.filename,
-            "total_size": total_size,
-            "parts": part_number,
-            "download_url": f"/dl/{unique_id}"
-        }
-    
-    except Exception as e:
-        logger.error(f"Upload failed: {str(e)}", exc_info=True)
-        raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")
-
-
-@app.get("/dl/{unique_id}")
-async def stream_file(unique_id: str, request: Request):
-    """
-    High-speed streaming endpoint with full range request support
-    Supports multi-part concatenation and parallel connections
-    """
-    if not session_manager or not database:
-        raise HTTPException(status_code=503, detail="Service not initialized")
-    
-    # Fetch file metadata
-    metadata = await database.get_file_metadata(unique_id)
-    if not metadata:
-        raise HTTPException(status_code=404, detail="File not found")
-    
-    # Parse range header
-    range_header = request.headers.get("range")
-    start = 0
-    end = metadata.total_size - 1
-    status_code = 200
-    
-    if range_header:
-        # Parse range: bytes=start-end
-        range_str = range_header.replace("bytes=", "")
-        range_parts = range_str.split("-")
-        
-        if range_parts[0]:
-            start = int(range_parts[0])
-        if range_parts[1]:
-            end = int(range_parts[1])
-        
-        status_code = 206  # Partial Content
-    
-    # Validate range
-    if start < 0 or end >= metadata.total_size or start > end:
-        raise HTTPException(status_code=416, detail="Range not satisfiable")
-    
-    logger.info(
-        f"Streaming request: unique_id={unique_id}, "
-        f"range={start}-{end}, size={end - start + 1}"
-    )
-    
-    # Create streaming response
-    content_length = end - start + 1
-    
-    headers = {
-        "Content-Type": "application/octet-stream",
-        "Content-Length": str(content_length),
-        "Accept-Ranges": "bytes",
-        "Content-Disposition": f'attachment; filename="{metadata.filename}"',
-    }
-    
-    if status_code == 206:
-        headers["Content-Range"] = f"bytes {start}-{end}/{metadata.total_size}"
-    
-    async def stream_generator() -> AsyncGenerator[bytes, None]:
-        """Generate stream from Telegram parts"""
-        bytes_sent = 0
-        current_position = 0
-        
-        for part in metadata.parts:
-            part_start = current_position
-            part_end = current_position + part["size"] - 1
-            
-            # Check if this part overlaps with requested range
-            if part_end < start:
-                current_position += part["size"]
-                continue
-            
-            if part_start > end:
-                break
-            
-            # Calculate offset within this part
-            offset_in_part = max(0, start - part_start)
-            bytes_to_read = min(
-                part["size"] - offset_in_part,
-                content_length - bytes_sent
-            )
-            
-            logger.debug(
-                f"Streaming part {part['part_number']}: "
-                f"offset={offset_in_part}, bytes={bytes_to_read}"
-            )
-            
-            # Stream this part with retry logic
-            retry_count = 0
-            max_retries = 3
-            
-            while retry_count < max_retries:
-                try:
-                    async for chunk in session_manager.stream_part(
-                        part["file_id"],
-                        offset=offset_in_part,
-                        limit=bytes_to_read
-                    ):
-                        chunk_size = len(chunk)
-                        
-                        # Ensure we don't send more than requested
-                        if bytes_sent + chunk_size > content_length:
-                            chunk = chunk[:content_length - bytes_sent]
-                            chunk_size = len(chunk)
-                        
-                        yield chunk
-                        bytes_sent += chunk_size
-                        
-                        if bytes_sent >= content_length:
-                            return
-                    
-                    break  # Success
-                
-                except Exception as e:
-                    retry_count += 1
-                    if retry_count >= max_retries:
-                        logger.error(
-                            f"Failed to stream part {part['part_number']}: {str(e)}"
-                        )
-                        raise
-                    
-                    wait_time = 2 ** retry_count
-                    logger.warning(
-                        f"Retry {retry_count}/{max_retries} for part "
-                        f"{part['part_number']} after {wait_time}s"
-                    )
-                    await asyncio.sleep(wait_time)
-            
-            current_position += part["size"]
-    
-    return StreamingResponse(
-        stream_generator(),
-        status_code=status_code,
-        headers=headers,
-        media_type="application/octet-stream"
-    )
-
-
-@app.get("/info/{unique_id}")
-async def get_file_info(unique_id: str):
-    """Get file metadata and information"""
-    if not database:
-        raise HTTPException(status_code=503, detail="Service not initialized")
-    
-    metadata = await database.get_file_metadata(unique_id)
-    if not metadata:
-        raise HTTPException(status_code=404, detail="File not found")
-    
-    return {
-        "unique_id": metadata.unique_id,
-        "filename": metadata.filename,
-        "total_size": metadata.total_size,
-        "total_size_gb": f"{metadata.total_size / (1024**3):.2f}",
-        "parts": metadata.part_count,
-        "uploaded_at": metadata.uploaded_at,
-        "download_url": f"/dl/{unique_id}"
-    }
-
-
-@app.delete("/delete/{unique_id}")
-async def delete_file(unique_id: str):
-    """Delete file and all its parts"""
-    if not session_manager or not database:
-        raise HTTPException(status_code=503, detail="Service not initialized")
-    
-    # Get metadata
-    metadata = await database.get_file_metadata(unique_id)
-    if not metadata:
-        raise HTTPException(status_code=404, detail="File not found")
-    
-    # Delete from Telegram (best effort)
-    deleted_parts = 0
-    for part in metadata.parts:
-        try:
-            await session_manager.delete_part(part["file_id"])
-            deleted_parts += 1
-        except Exception as e:
-            logger.warning(f"Failed to delete part {part['part_number']}: {str(e)}")
-    
-    # Delete from database
-    await database.delete_file_metadata(unique_id)
-    
-    logger.info(f"Deleted file: unique_id={unique_id}, parts={deleted_parts}")
-    
-    return {
-        "success": True,
-        "unique_id": unique_id,
-        "deleted_parts": deleted_parts,
-        "total_parts": metadata.part_count
-    }
-
-
-if __name__ == "__main__":
-    uvicorn.run(
-        "main:app",
-        host="0.0.0.0",
-        port=8000,
-        workers=1,  # Single worker for shared session state
-        log_level="info"
-    )

files/mnt/user-data/outputs/telegram-streamer/database.py

@@ -1,199 +0,0 @@
-"""
-Database Module - MongoDB schema and operations for file metadata
-"""
-import logging
-from typing import List, Optional, Dict, Any
-from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
-from datetime import datetime
-
-logger = logging.getLogger(__name__)
-
-
-class Database:
-    """MongoDB database manager for file metadata"""
-    
-    def __init__(self, mongo_uri: str):
-        self.client: AsyncIOMotorClient = AsyncIOMotorClient(mongo_uri)
-        self.db: AsyncIOMotorDatabase = self.client.telegram_streamer
-        self.files_collection = self.db.files
-        
-    async def initialize(self):
-        """Create indexes for optimal query performance"""
-        try:
-            await self.files_collection.create_index("unique_id", unique=True)
-            await self.files_collection.create_index("created_at")
-            logger.info("Database indexes created successfully")
-        except Exception as e:
-            logger.error(f"Error creating indexes: {e}")
-    
-    async def create_file_metadata(
-        self,
-        unique_id: str,
-        filename: str,
-        total_size: int,
-        part_size: int,
-        mime_type: str = "application/octet-stream"
-    ) -> Dict[str, Any]:
-        """
-        Create initial file metadata entry
-        
-        Args:
-            unique_id: Unique identifier for the file
-            filename: Original filename
-            total_size: Total file size in bytes
-            part_size: Size of each part (except last)
-            mime_type: MIME type of the file
-            
-        Returns:
-            Created document
-        """
-        document = {
-            "unique_id": unique_id,
-            "filename": filename,
-            "total_size": total_size,
-            "part_size": part_size,
-            "mime_type": mime_type,
-            "parts": [],
-            "total_parts": 0,
-            "upload_status": "in_progress",
-            "created_at": datetime.utcnow(),
-            "updated_at": datetime.utcnow()
-        }
-        
-        await self.files_collection.insert_one(document)
-        logger.info(f"Created metadata for {unique_id}: {filename} ({total_size} bytes)")
-        return document
-    
-    async def add_file_part(
-        self,
-        unique_id: str,
-        part_number: int,
-        file_id: str,
-        part_size: int
-    ) -> bool:
-        """
-        Add a file part to the metadata
-        
-        Args:
-            unique_id: Unique identifier for the file
-            part_number: Part sequence number (0-indexed)
-            file_id: Telegram file_id for this part
-            part_size: Actual size of this part in bytes
-            
-        Returns:
-            Success status
-        """
-        try:
-            result = await self.files_collection.update_one(
-                {"unique_id": unique_id},
-                {
-                    "$push": {
-                        "parts": {
-                            "part_number": part_number,
-                            "file_id": file_id,
-                            "size": part_size
-                        }
-                    },
-                    "$inc": {"total_parts": 1},
-                    "$set": {"updated_at": datetime.utcnow()}
-                }
-            )
-            
-            if result.modified_count > 0:
-                logger.info(f"Added part {part_number} to {unique_id}")
-                return True
-            return False
-        except Exception as e:
-            logger.error(f"Error adding part {part_number} to {unique_id}: {e}")
-            return False
-    
-    async def complete_upload(self, unique_id: str) -> bool:
-        """Mark upload as complete"""
-        try:
-            result = await self.files_collection.update_one(
-                {"unique_id": unique_id},
-                {
-                    "$set": {
-                        "upload_status": "completed",
-                        "updated_at": datetime.utcnow()
-                    }
-                }
-            )
-            
-            if result.modified_count > 0:
-                logger.info(f"Upload completed for {unique_id}")
-                return True
-            return False
-        except Exception as e:
-            logger.error(f"Error completing upload for {unique_id}: {e}")
-            return False
-    
-    async def mark_upload_failed(self, unique_id: str, error: str) -> bool:
-        """Mark upload as failed"""
-        try:
-            result = await self.files_collection.update_one(
-                {"unique_id": unique_id},
-                {
-                    "$set": {
-                        "upload_status": "failed",
-                        "error": error,
-                        "updated_at": datetime.utcnow()
-                    }
-                }
-            )
-            
-            if result.modified_count > 0:
-                logger.info(f"Upload failed for {unique_id}: {error}")
-                return True
-            return False
-        except Exception as e:
-            logger.error(f"Error marking upload as failed for {unique_id}: {e}")
-            return False
-    
-    async def get_file_metadata(self, unique_id: str) -> Optional[Dict[str, Any]]:
-        """
-        Retrieve file metadata by unique_id
-        
-        Args:
-            unique_id: Unique identifier for the file
-            
-        Returns:
-            File metadata document or None if not found
-        """
-        try:
-            doc = await self.files_collection.find_one({"unique_id": unique_id})
-            if doc:
-                # Sort parts by part_number for ordered retrieval
-                if "parts" in doc and doc["parts"]:
-                    doc["parts"] = sorted(doc["parts"], key=lambda x: x["part_number"])
-            return doc
-        except Exception as e:
-            logger.error(f"Error retrieving metadata for {unique_id}: {e}")
-            return None
-    
-    async def delete_file_metadata(self, unique_id: str) -> bool:
-        """Delete file metadata"""
-        try:
-            result = await self.files_collection.delete_one({"unique_id": unique_id})
-            if result.deleted_count > 0:
-                logger.info(f"Deleted metadata for {unique_id}")
-                return True
-            return False
-        except Exception as e:
-            logger.error(f"Error deleting metadata for {unique_id}: {e}")
-            return False
-    
-    async def get_all_files(self, limit: int = 100, skip: int = 0) -> List[Dict[str, Any]]:
-        """Get list of all files with pagination"""
-        try:
-            cursor = self.files_collection.find().sort("created_at", -1).skip(skip).limit(limit)
-            files = await cursor.to_list(length=limit)
-            return files
-        except Exception as e:
-            logger.error(f"Error retrieving file list: {e}")
-            return []
-    
-    async def close(self):
-        """Close database connection"""
-        self.client.close()
-        logger.info("Database connection closed")

files/requirements.txt

@@ -1,20 +0,0 @@
-# Core Framework
-fastapi==0.109.0
-uvicorn[standard]==0.27.0
-python-multipart==0.0.6
-
-# Telegram Client
-pyrogram==2.0.106
-TgCrypto==1.2.5
-
-# Database
-motor==3.3.2
-pymongo==4.6.1
-
-# Utilities
-python-dotenv==1.0.0
-aiofiles==23.2.1
-
-# Performance
-uvloop==0.19.0
-httptools==0.6.1

files/session_manager.py

@@ -1,338 +0,0 @@
-"""
-Session Manager - Multi-session rotation and load balancing
-Handles multiple Pyrogram sessions to maximize bandwidth and avoid flood limits
-"""
-
-import asyncio
-import logging
-import os
-from typing import List, Optional, AsyncGenerator
-from io import BytesIO
-
-from pyrogram import Client
-from pyrogram.errors import FloodWait, BadRequest
-from pyrogram.types import Message
-
-from utils import CHUNK_SIZE
-
-logger = logging.getLogger(__name__)
-
-
-class TelegramSession:
-    """Wrapper for a single Pyrogram session"""
-    
-    def __init__(
-        self,
-        session_name: str,
-        api_id: int,
-        api_hash: str,
-        session_string: Optional[str] = None
-    ):
-        self.session_name = session_name
-        self.api_id = api_id
-        self.api_hash = api_hash
-        self.session_string = session_string
-        self.client: Optional[Client] = None
-        self.is_active = False
-        self.upload_count = 0
-        self.download_count = 0
-        
-    async def initialize(self):
-        """Initialize and start the Pyrogram client"""
-        try:
-            if self.session_string:
-                self.client = Client(
-                    name=self.session_name,
-                    api_id=self.api_id,
-                    api_hash=self.api_hash,
-                    session_string=self.session_string,
-                    in_memory=True,
-                    no_updates=True
-                )
-            else:
-                self.client = Client(
-                    name=self.session_name,
-                    api_id=self.api_id,
-                    api_hash=self.api_hash,
-                    in_memory=True,
-                    no_updates=True
-                )
-            
-            await self.client.start()
-            self.is_active = True
-            
-            me = await self.client.get_me()
-            logger.info(
-                f"Session {self.session_name} initialized: "
-                f"@{me.username or me.first_name}"
-            )
-            
-        except Exception as e:
-            logger.error(f"Failed to initialize session {self.session_name}: {str(e)}")
-            self.is_active = False
-            raise
-    
-    async def cleanup(self):
-        """Stop and cleanup the session"""
-        if self.client and self.is_active:
-            try:
-                await self.client.stop()
-                self.is_active = False
-                logger.info(f"Session {self.session_name} stopped")
-            except Exception as e:
-                logger.error(f"Error stopping session {self.session_name}: {str(e)}")
-
-
-class SessionManager:
-    """Manages multiple Telegram sessions for load balancing"""
-    
-    def __init__(self):
-        self.sessions: List[TelegramSession] = []
-        self.bot_token: Optional[str] = None
-        self.bot_session: Optional[TelegramSession] = None
-        self.current_upload_index = 0
-        self.current_download_index = 0
-        self.lock = asyncio.Lock()
-    
-    async def initialize(self):
-        """Initialize all sessions from environment variables"""
-        logger.info("Initializing Session Manager...")
-        
-        # Get bot token
-        self.bot_token = os.getenv("BOT_TOKEN")
-        if not self.bot_token:
-            raise ValueError("BOT_TOKEN environment variable is required")
-        
-        # Initialize bot session
-        api_id = int(os.getenv("API_ID", "0"))
-        api_hash = os.getenv("API_HASH", "")
-        
-        if not api_id or not api_hash:
-            raise ValueError("API_ID and API_HASH environment variables are required")
-        
-        self.bot_session = TelegramSession(
-            session_name="bot_session",
-            api_id=api_id,
-            api_hash=api_hash
-        )
-        
-        # Override with bot token
-        self.bot_session.client = Client(
-            name="bot_session",
-            api_id=api_id,
-            api_hash=api_hash,
-            bot_token=self.bot_token,
-            in_memory=True,
-            no_updates=True
-        )
-        
-        await self.bot_session.client.start()
-        self.bot_session.is_active = True
-        logger.info("Bot session initialized")
-        
-        # Initialize user sessions from SESSION_STRINGS
-        session_strings = os.getenv("SESSION_STRINGS", "").split(",")
-        session_strings = [s.strip() for s in session_strings if s.strip()]
-        
-        if not session_strings:
-            logger.warning("No SESSION_STRINGS found, using bot session only")
-            self.sessions = [self.bot_session]
-            return
-        
-        # Create user sessions
-        for i, session_string in enumerate(session_strings):
-            session = TelegramSession(
-                session_name=f"user_session_{i}",
-                api_id=api_id,
-                api_hash=api_hash,
-                session_string=session_string
-            )
-            
-            try:
-                await session.initialize()
-                self.sessions.append(session)
-            except Exception as e:
-                logger.error(f"Failed to initialize user session {i}: {str(e)}")
-        
-        if not self.sessions:
-            # Fallback to bot session
-            self.sessions = [self.bot_session]
-        
-        logger.info(f"Session Manager initialized with {len(self.sessions)} session(s)")
-    
-    def get_next_upload_session(self) -> TelegramSession:
-        """Get next session for upload (round-robin)"""
-        if not self.sessions:
-            raise RuntimeError("No active sessions available")
-        
-        session = self.sessions[self.current_upload_index]
-        self.current_upload_index = (self.current_upload_index + 1) % len(self.sessions)
-        
-        return session
-    
-    def get_next_download_session(self) -> TelegramSession:
-        """Get next session for download (round-robin)"""
-        if not self.sessions:
-            raise RuntimeError("No active sessions available")
-        
-        session = self.sessions[self.current_download_index]
-        self.current_download_index = (
-            self.current_download_index + 1
-        ) % len(self.sessions)
-        
-        return session
-    
-    async def upload_part(
-        self,
-        data: bytes,
-        filename: str,
-        max_retries: int = 3
-    ) -> str:
-        """
-        Upload a file part to Telegram
-        Returns: file_id for later retrieval
-        """
-        retry_count = 0
-        
-        while retry_count < max_retries:
-            session = self.get_next_upload_session()
-            
-            if not session.is_active or not session.client:
-                retry_count += 1
-                continue
-            
-            try:
-                # Upload to "Saved Messages" (self chat)
-                message: Message = await session.client.send_document(
-                    chat_id="me",
-                    document=BytesIO(data),
-                    file_name=filename,
-                    force_document=True
-                )
-                
-                session.upload_count += 1
-                file_id = message.document.file_id
-                
-                logger.debug(
-                    f"Part uploaded via {session.session_name}: "
-                    f"file_id={file_id}, size={len(data)}"
-                )
-                
-                return file_id
-            
-            except FloodWait as e:
-                logger.warning(
-                    f"FloodWait on {session.session_name}: waiting {e.value}s"
-                )
-                await asyncio.sleep(e.value)
-                retry_count += 1
-            
-            except Exception as e:
-                logger.error(
-                    f"Upload failed on {session.session_name}: {str(e)}"
-                )
-                retry_count += 1
-                
-                if retry_count < max_retries:
-                    await asyncio.sleep(2 ** retry_count)
-        
-        raise RuntimeError(f"Failed to upload part after {max_retries} retries")
-    
-    async def stream_part(
-        self,
-        file_id: str,
-        offset: int = 0,
-        limit: Optional[int] = None
-    ) -> AsyncGenerator[bytes, None]:
-        """
-        Stream a file part from Telegram
-        Yields chunks of data
-        """
-        session = self.get_next_download_session()
-        
-        if not session.is_active or not session.client:
-            raise RuntimeError("No active session available for streaming")
-        
-        try:
-            bytes_read = 0
-            
-            async for chunk in session.client.stream_media(
-                file_id,
-                offset=offset,
-                limit=limit or 0
-            ):
-                if limit and bytes_read + len(chunk) > limit:
-                    # Trim final chunk
-                    yield chunk[:limit - bytes_read]
-                    break
-                
-                yield chunk
-                bytes_read += len(chunk)
-                
-                if limit and bytes_read >= limit:
-                    break
-            
-            session.download_count += 1
-            
-            logger.debug(
-                f"Part streamed via {session.session_name}: "
-                f"file_id={file_id}, bytes={bytes_read}"
-            )
-        
-        except FloodWait as e:
-            logger.warning(f"FloodWait on download: waiting {e.value}s")
-            await asyncio.sleep(e.value)
-            
-            # Retry with next session
-            async for chunk in self.stream_part(file_id, offset, limit):
-                yield chunk
-        
-        except Exception as e:
-            logger.error(f"Stream failed: {str(e)}")
-            raise
-    
-    async def delete_part(self, file_id: str) -> bool:
-        """Delete a file part from Telegram"""
-        session = self.get_next_upload_session()
-        
-        if not session.is_active or not session.client:
-            return False
-        
-        try:
-            # Get message and delete it
-            # Note: This requires the message_id, which we don't store
-            # For production, consider storing message_ids in metadata
-            logger.warning("Delete operation requires message_id, not implemented")
-            return False
-        
-        except Exception as e:
-            logger.error(f"Delete failed: {str(e)}")
-            return False
-    
-    async def cleanup(self):
-        """Cleanup all sessions"""
-        logger.info("Cleaning up Session Manager...")
-        
-        for session in self.sessions:
-            await session.cleanup()
-        
-        if self.bot_session and self.bot_session != self.sessions[0]:
-            await self.bot_session.cleanup()
-        
-        logger.info("Session Manager cleanup complete")
-    
-    def get_stats(self) -> dict:
-        """Get session statistics"""
-        return {
-            "total_sessions": len(self.sessions),
-            "active_sessions": sum(1 for s in self.sessions if s.is_active),
-            "sessions": [
-                {
-                    "name": s.session_name,
-                    "active": s.is_active,
-                    "uploads": s.upload_count,
-                    "downloads": s.download_count
-                }
-                for s in self.sessions
-            ]
-        }

files/test_setup.py

@@ -1,325 +0,0 @@
-#!/usr/bin/env python3
-"""
-Test Script for Telegram Multi-Part File Streamer
-Validates setup and performs basic functionality tests
-"""
-
-import asyncio
-import os
-import sys
-import time
-from io import BytesIO
-
-import httpx
-from dotenv import load_dotenv
-
-
-# Load environment variables
-load_dotenv()
-
-
-class Colors:
-    """ANSI color codes"""
-    GREEN = '\033[92m'
-    YELLOW = '\033[93m'
-    RED = '\033[91m'
-    BLUE = '\033[94m'
-    ENDC = '\033[0m'
-    BOLD = '\033[1m'
-
-
-def print_header(text: str):
-    """Print section header"""
-    print(f"\n{Colors.BOLD}{Colors.BLUE}{'=' * 60}{Colors.ENDC}")
-    print(f"{Colors.BOLD}{Colors.BLUE}{text.center(60)}{Colors.ENDC}")
-    print(f"{Colors.BOLD}{Colors.BLUE}{'=' * 60}{Colors.ENDC}\n")
-
-
-def print_success(text: str):
-    """Print success message"""
-    print(f"{Colors.GREEN}✓ {text}{Colors.ENDC}")
-
-
-def print_error(text: str):
-    """Print error message"""
-    print(f"{Colors.RED}✗ {text}{Colors.ENDC}")
-
-
-def print_warning(text: str):
-    """Print warning message"""
-    print(f"{Colors.YELLOW}⚠ {text}{Colors.ENDC}")
-
-
-def print_info(text: str):
-    """Print info message"""
-    print(f"{Colors.BLUE}ℹ {text}{Colors.ENDC}")
-
-
-async def test_environment_variables():
-    """Test 1: Check environment variables"""
-    print_header("Test 1: Environment Variables")
-    
-    required_vars = {
-        "API_ID": "Telegram API ID",
-        "API_HASH": "Telegram API Hash",
-        "BOT_TOKEN": "Telegram Bot Token",
-        "MONGO_URI": "MongoDB Connection String"
-    }
-    
-    optional_vars = {
-        "SESSION_STRINGS": "Pyrogram Session Strings (for load balancing)"
-    }
-    
-    all_valid = True
-    
-    print_info("Checking required variables...")
-    for var, description in required_vars.items():
-        value = os.getenv(var)
-        if value:
-            masked_value = value[:8] + "..." if len(value) > 8 else value
-            print_success(f"{var}: {masked_value} ({description})")
-        else:
-            print_error(f"{var}: Missing! ({description})")
-            all_valid = False
-    
-    print()
-    print_info("Checking optional variables...")
-    for var, description in optional_vars.items():
-        value = os.getenv(var)
-        if value:
-            count = len(value.split(","))
-            print_success(f"{var}: {count} session(s) configured ({description})")
-        else:
-            print_warning(f"{var}: Not set ({description})")
-    
-    print()
-    return all_valid
-
-
-async def test_mongodb_connection():
-    """Test 2: MongoDB connection"""
-    print_header("Test 2: MongoDB Connection")
-    
-    try:
-        from motor.motor_asyncio import AsyncIOMotorClient
-        
-        mongo_uri = os.getenv("MONGO_URI")
-        if not mongo_uri:
-            print_error("MONGO_URI not set")
-            return False
-        
-        print_info(f"Connecting to MongoDB...")
-        client = AsyncIOMotorClient(mongo_uri, serverSelectionTimeoutMS=5000)
-        
-        # Test connection
-        await client.admin.command('ping')
-        
-        # Get database info
-        db_name = os.getenv("MONGO_DATABASE", "telegram_streamer")
-        db = client[db_name]
-        
-        print_success(f"Connected to MongoDB!")
-        print_info(f"Database: {db_name}")
-        
-        # List collections
-        collections = await db.list_collection_names()
-        print_info(f"Collections: {collections if collections else 'None (fresh database)'}")
-        
-        client.close()
-        return True
-        
-    except Exception as e:
-        print_error(f"MongoDB connection failed: {str(e)}")
-        return False
-
-
-async def test_api_server():
-    """Test 3: API server availability"""
-    print_header("Test 3: API Server")
-    
-    base_url = "http://localhost:8000"
-    
-    print_info(f"Testing API server at {base_url}...")
-    
-    try:
-        async with httpx.AsyncClient(timeout=10.0) as client:
-            # Test root endpoint
-            response = await client.get(base_url)
-            if response.status_code == 200:
-                data = response.json()
-                print_success(f"Root endpoint: {data.get('status', 'unknown')}")
-            else:
-                print_error(f"Root endpoint returned {response.status_code}")
-                return False
-            
-            # Test health endpoint
-            response = await client.get(f"{base_url}/health")
-            if response.status_code == 200:
-                data = response.json()
-                print_success(f"Health check: {data.get('status', 'unknown')}")
-                print_info(f"Active sessions: {data.get('sessions', 0)}")
-                print_info(f"Database: {data.get('database', 'unknown')}")
-            else:
-                print_warning(f"Health endpoint returned {response.status_code}")
-            
-            return True
-            
-    except httpx.ConnectError:
-        print_error("Cannot connect to API server!")
-        print_info("Make sure the server is running:")
-        print_info("  uvicorn main:app --reload")
-        return False
-    except Exception as e:
-        print_error(f"API test failed: {str(e)}")
-        return False
-
-
-async def test_upload_download():
-    """Test 4: Upload and download functionality"""
-    print_header("Test 4: Upload/Download")
-    
-    base_url = "http://localhost:8000"
-    
-    print_info("Creating test file (1MB)...")
-    test_data = os.urandom(1024 * 1024)  # 1MB random data
-    test_filename = "test_file.bin"
-    
-    try:
-        async with httpx.AsyncClient(timeout=60.0) as client:
-            # Upload
-            print_info("Uploading test file...")
-            start_time = time.time()
-            
-            response = await client.post(
-                f"{base_url}/upload",
-                params={"filename": test_filename},
-                content=test_data
-            )
-            
-            upload_time = time.time() - start_time
-            
-            if response.status_code != 200:
-                print_error(f"Upload failed: {response.status_code}")
-                print_error(f"Response: {response.text}")
-                return False
-            
-            result = response.json()
-            unique_id = result.get("unique_id")
-            
-            print_success(f"Upload completed in {upload_time:.2f}s")
-            print_info(f"Unique ID: {unique_id}")
-            print_info(f"Parts: {result.get('parts', 0)}")
-            print_info(f"Size: {result.get('total_size', 0)} bytes")
-            
-            # Download
-            print_info("Downloading test file...")
-            start_time = time.time()
-            
-            response = await client.get(f"{base_url}/dl/{unique_id}")
-            download_time = time.time() - start_time
-            
-            if response.status_code != 200:
-                print_error(f"Download failed: {response.status_code}")
-                return False
-            
-            downloaded_data = response.content
-            
-            print_success(f"Download completed in {download_time:.2f}s")
-            
-            # Verify
-            if downloaded_data == test_data:
-                print_success("Data integrity verified! ✓")
-            else:
-                print_error("Data integrity check failed!")
-                return False
-            
-            # Test range request
-            print_info("Testing range request (bytes 0-1023)...")
-            response = await client.get(
-                f"{base_url}/dl/{unique_id}",
-                headers={"Range": "bytes=0-1023"}
-            )
-            
-            if response.status_code == 206:
-                print_success("Range request supported! ✓")
-                if len(response.content) == 1024:
-                    print_success("Range request data correct! ✓")
-                else:
-                    print_error(f"Range request returned {len(response.content)} bytes, expected 1024")
-            else:
-                print_warning(f"Range request returned {response.status_code} (expected 206)")
-            
-            # Cleanup
-            print_info("Cleaning up test file...")
-            response = await client.delete(f"{base_url}/delete/{unique_id}")
-            if response.status_code == 200:
-                print_success("Test file deleted")
-            
-            return True
-            
-    except Exception as e:
-        print_error(f"Upload/Download test failed: {str(e)}")
-        return False
-
-
-async def main():
-    """Run all tests"""
-    print()
-    print(f"{Colors.BOLD}{Colors.BLUE}")
-    print("╔════════════════════════════════════════════════════════════╗")
-    print("║  Telegram Multi-Part File Streamer - Test Suite           ║")
-    print("╚════════════════════════════════════════════════════════════╝")
-    print(f"{Colors.ENDC}")
-    
-    tests = [
-        ("Environment Variables", test_environment_variables),
-        ("MongoDB Connection", test_mongodb_connection),
-        ("API Server", test_api_server),
-        ("Upload/Download", test_upload_download)
-    ]
-    
-    results = []
-    
-    for name, test_func in tests:
-        try:
-            result = await test_func()
-            results.append((name, result))
-        except Exception as e:
-            print_error(f"Test '{name}' crashed: {str(e)}")
-            results.append((name, False))
-    
-    # Summary
-    print_header("Test Summary")
-    
-    passed = sum(1 for _, result in results if result)
-    total = len(results)
-    
-    for name, result in results:
-        if result:
-            print_success(f"{name}")
-        else:
-            print_error(f"{name}")
-    
-    print()
-    if passed == total:
-        print_success(f"All tests passed! ({passed}/{total})")
-        print()
-        print_info("Your setup is ready! 🚀")
-        print_info("You can now start uploading and streaming files.")
-        print()
-        return 0
-    else:
-        print_error(f"Some tests failed: {passed}/{total} passed")
-        print()
-        print_info("Please fix the issues above before proceeding.")
-        print()
-        return 1
-
-
-if __name__ == "__main__":
-    try:
-        exit_code = asyncio.run(main())
-        sys.exit(exit_code)
-    except KeyboardInterrupt:
-        print(f"\n\n{Colors.YELLOW}Tests cancelled by user.{Colors.ENDC}\n")
-        sys.exit(1)

files/utils.py

@@ -1,162 +0,0 @@
-"""
-Utility Functions and Constants
-Helper functions for file streaming and chunk management
-"""
-
-import uuid
-from typing import Tuple
-
-# Constants
-CHUNK_SIZE = 1024 * 1024  # 1MB chunks for streaming
-MAX_PART_SIZE = 2000 * 1024 * 1024  # 2000MB (2GB) per part for safety
-TELEGRAM_FILE_LIMIT = 2048 * 1024 * 1024  # Telegram's 2GB limit
-
-
-def generate_unique_id() -> str:
-    """Generate a unique identifier for files"""
-    return uuid.uuid4().hex[:16]
-
-
-def calculate_part_and_offset(
-    byte_position: int,
-    parts_info: list
-) -> Tuple[int, int, int]:
-    """
-    Calculate which Telegram part contains a byte position
-    and the offset within that part
-    
-    Args:
-        byte_position: Absolute byte position in the file
-        parts_info: List of dicts with 'part_number', 'size', 'file_id'
-    
-    Returns:
-        Tuple of (part_index, offset_in_part, part_file_id)
-    """
-    current_position = 0
-    
-    for i, part in enumerate(parts_info):
-        part_start = current_position
-        part_end = current_position + part["size"]
-        
-        if byte_position >= part_start and byte_position < part_end:
-            offset = byte_position - part_start
-            return (i, offset, part["file_id"])
-        
-        current_position += part["size"]
-    
-    # If we reach here, position is beyond file size
-    raise ValueError(f"Byte position {byte_position} exceeds file size")
-
-
-def format_size(size_bytes: int) -> str:
-    """Format byte size to human-readable string"""
-    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
-        if size_bytes < 1024.0:
-            return f"{size_bytes:.2f} {unit}"
-        size_bytes /= 1024.0
-    return f"{size_bytes:.2f} PB"
-
-
-def validate_range(start: int, end: int, total_size: int) -> bool:
-    """Validate HTTP range request parameters"""
-    if start < 0 or end < 0:
-        return False
-    if start > end:
-        return False
-    if end >= total_size:
-        return False
-    return True
-
-
-def split_into_parts(total_size: int, part_size: int = MAX_PART_SIZE) -> list:
-    """
-    Calculate how a file should be split into parts
-    
-    Args:
-        total_size: Total file size in bytes
-        part_size: Maximum size per part
-    
-    Returns:
-        List of tuples (part_number, start_byte, end_byte)
-    """
-    parts = []
-    current_position = 0
-    part_number = 1
-    
-    while current_position < total_size:
-        end_position = min(current_position + part_size, total_size)
-        parts.append((part_number, current_position, end_position))
-        current_position = end_position
-        part_number += 1
-    
-    return parts
-
-
-class CircularBuffer:
-    """Circular buffer for efficient memory management"""
-    
-    def __init__(self, size: int):
-        self.size = size
-        self.buffer = bytearray(size)
-        self.write_pos = 0
-        self.read_pos = 0
-        self.available = 0
-    
-    def write(self, data: bytes) -> int:
-        """Write data to buffer, returns bytes written"""
-        space = self.size - self.available
-        to_write = min(len(data), space)
-        
-        if to_write == 0:
-            return 0
-        
-        # Handle wrap-around
-        end_pos = self.write_pos + to_write
-        if end_pos <= self.size:
-            self.buffer[self.write_pos:end_pos] = data[:to_write]
-        else:
-            first_part = self.size - self.write_pos
-            self.buffer[self.write_pos:] = data[:first_part]
-            self.buffer[:to_write - first_part] = data[first_part:to_write]
-        
-        self.write_pos = (self.write_pos + to_write) % self.size
-        self.available += to_write
-        
-        return to_write
-    
-    def read(self, length: int) -> bytes:
-        """Read data from buffer"""
-        to_read = min(length, self.available)
-        
-        if to_read == 0:
-            return b''
-        
-        # Handle wrap-around
-        end_pos = self.read_pos + to_read
-        if end_pos <= self.size:
-            data = bytes(self.buffer[self.read_pos:end_pos])
-        else:
-            first_part = self.size - self.read_pos
-            data = (
-                bytes(self.buffer[self.read_pos:]) +
-                bytes(self.buffer[:to_read - first_part])
-            )
-        
-        self.read_pos = (self.read_pos + to_read) % self.size
-        self.available -= to_read
-        
-        return data
-    
-    def is_full(self) -> bool:
-        """Check if buffer is full"""
-        return self.available == self.size
-    
-    def is_empty(self) -> bool:
-        """Check if buffer is empty"""
-        return self.available == 0
-    
-    def clear(self):
-        """Clear the buffer"""
-        self.write_pos = 0
-        self.read_pos = 0
-        self.available = 0