chronicle.nixg.ru/telegram-archiver/test_full_archive.py

#!/usr/bin/env python
"""Test full channel archive with flood handling."""

import asyncio
import os
import sys
from pathlib import Path

project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
os.chdir(project_root)

from app.telethon_client import TelethonArchiver
from app.archiver import ChannelArchiver
from config import get_settings
import logging

logging.basicConfig(level=logging.INFO)

async def main():
    settings = get_settings()
    
    print("=" * 60)
    print("FULL CHANNEL ARCHIVE TEST")
    print("=" * 60)
    
    async with TelethonArchiver(settings) as client:
        # Get channel info
        entity = await client.get_channel_info('dedinit')
        print(f"Channel: {entity.title}")
        print("=" * 60)
        
        # Archive all
        archiver = ChannelArchiver(client, settings)
        result = await archiver.archive_channel(
            channel='dedinit',
            limit=None,  # All posts
            force=False  # Skip existing
        )
        
        print("\n" + "=" * 60)
        print("RESULTS")
        print("=" * 60)
        print(f"Posts archived: {result['posts_archived']}")
        print(f"Posts skipped: {result['posts_skipped']}")
        print(f"Duration: {result['duration_seconds']:.2f}s")
        print(f"Output: {result['output_path']}")
        print("=" * 60)

if __name__ == "__main__":
    import os
    asyncio.run(main())