import logging import os import mimetypes import asyncio import json from datetime import datetime from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Response, BackgroundTasks from fastapi.responses import HTMLResponse, FileResponse, PlainTextResponse from telegram_client import TelegramClient from config import get_settings from rss_generator import generate_channel_rss from pyrogram import errors from post_parser import PostParser logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) if not logger.handlers: handler = logging.StreamHandler() handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) logger.addHandler(handler) client = TelegramClient() Config = get_settings() @asynccontextmanager async def lifespan(_: FastAPI): await client.start() # Start background task background_task = asyncio.create_task(cache_media_files()) yield # Cleanup background_task.cancel() try: await background_task except asyncio.CancelledError: pass await client.stop() app = FastAPI( title="Pyrogram Bridge", lifespan=lifespan) if __name__ == "__main__": import uvicorn uvicorn.run("api_server:app", host=Config["api_host"], port=Config["api_port"], reload=True) async def find_file_id_in_message(message, file_unique_id: str): """Find file_id by checking all possible media types in message""" if message.photo and message.photo.file_unique_id == file_unique_id: return message.photo.file_id elif message.video and message.video.file_unique_id == file_unique_id: return message.video.file_id elif message.animation and message.animation.file_unique_id == file_unique_id: return message.animation.file_id elif message.video_note and message.video_note.file_unique_id == file_unique_id: return message.video_note.file_id elif message.audio and message.audio.file_unique_id == file_unique_id: return message.audio.file_id elif message.voice and message.voice.file_unique_id == file_unique_id: return message.voice.file_id elif message.sticker and message.sticker.file_unique_id == file_unique_id: return message.sticker.file_id elif message.web_page and message.web_page.photo and message.web_page.photo.file_unique_id == file_unique_id: return message.web_page.photo.file_id elif message.document and message.document.file_unique_id == file_unique_id: return message.document.file_id return None async def prepare_file_response(file_path: str, background_tasks: BackgroundTasks): """Prepare file response with proper headers""" if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found") media_type, _ = mimetypes.guess_type(file_path) if not media_type: media_type = "application/octet-stream" headers = {"Content-Disposition": f"inline; filename={os.path.basename(file_path)}"} return FileResponse(path=file_path, media_type=media_type, headers=headers) @app.get("/html/{channel}/{post_id}", response_class=HTMLResponse) @app.get("/post/html/{channel}/{post_id}", response_class=HTMLResponse) async def get_post_html(channel: str, post_id: int): try: parser = PostParser(client.client) html_content = await parser.get_post(channel, post_id, 'html') if not html_content: raise HTTPException(status_code=404, detail="Post not found") return html_content except Exception as e: error_message = f"Failed to get HTML post for channel {channel}, post_id {post_id}: {str(e)}" logger.error(error_message) raise HTTPException(status_code=500, detail=error_message) from e @app.get("/json/{channel}/{post_id}") @app.get("/post/json/{channel}/{post_id}") async def get_post(channel: str, post_id: int): try: parser = PostParser(client.client) json_content = await parser.get_post(channel, post_id, 'json') if not json_content: raise HTTPException(status_code=404, detail="Post not found") return json_content except Exception as e: error_message = f"Failed to get JSON post for channel {channel}, post_id {post_id}: {str(e)}" logger.error(error_message) raise HTTPException(status_code=500, detail=error_message) from e @app.get("/health") async def health_check(): try: me = await client.client.get_me() return { "status": "ok", "tg_connected": client.client.is_connected, "tg_name": me.username, "tg_id": me.id, "tg_phone": me.phone_number, "tg_first_name": me.first_name, "tg_last_name": me.last_name, } except Exception as e: error_message = f"Failed to get health check: {str(e)}" logger.error(error_message) raise HTTPException(status_code=500, detail=error_message) from e async def download_media_file(channel: str, post_id: int, file_unique_id: str) -> str: """ Download media file from Telegram and save to cache Returns path to downloaded file """ cache_dir = os.path.abspath("./data/cache") cache_path = os.path.join(cache_dir, f"{channel}-{post_id}-{file_unique_id}") os.makedirs(cache_dir, exist_ok=True) if os.path.exists(cache_path): logger.info(f"Found cached media file: {cache_path}") return cache_path message = await client.client.get_messages(channel, post_id) file_id = await find_file_id_in_message(message, file_unique_id) if not file_id: logger.error(f"Media with file_unique_id {file_unique_id} not found in message {post_id}") raise HTTPException(status_code=404, detail="File not found in message") file_path = await client.client.download_media(file_id, file_name=cache_path) logger.info(f"Downloaded media file {file_unique_id} to {cache_path}") return file_path @app.get("/media/{channel}/{post_id}/{file_unique_id}") async def get_media(channel: str, post_id: int, file_unique_id: str, background_tasks: BackgroundTasks): try: file_path = await download_media_file(channel, post_id, file_unique_id) return await prepare_file_response(file_path, background_tasks) except HTTPException: raise except errors.RPCError as e: logger.error(f"Media request RPC error for file_unique_id {file_unique_id} in message {post_id}: {type(e).__name__} - {str(e)}") raise HTTPException(status_code=404, detail="File not found in Telegram") from e except Exception as e: error_message = f"Failed to get media for file_unique_id {file_unique_id} in message {post_id}: {str(e)}" logger.error(error_message) raise HTTPException(status_code=500, detail=error_message) from e @app.get("/rss/{channel}", response_class=Response) async def get_rss_feed(channel: str, limit: int = 20): try: rss_content = await generate_channel_rss(channel, client=client.client, limit=limit) return Response(content=rss_content, media_type="application/xml") except ValueError as e: error_message = f"Invalid parameters for RSS feed generation: {str(e)}" logger.error(error_message) raise HTTPException(status_code=400, detail=error_message) from e except Exception as e: error_message = f"Failed to generate RSS feed for channel {channel}: {str(e)}" logger.error(error_message) raise HTTPException(status_code=500, detail=error_message) from e async def cache_media_files(): """Background task to download media files to cache""" while True: try: file_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') if not os.path.exists(file_path): await asyncio.sleep(60) continue with open(file_path, 'r', encoding='utf-8') as f: media_files = json.load(f) for file_data in media_files: try: file_id = file_data['file_id'] if not file_id: continue # Parse file_id to get parameters channel, post_id_str, file_unique_id = file_id.split('-', 2) post_id = int(post_id_str) # Convert post_id to integer # Check if file already exists in cache cache_dir = os.path.abspath("./data/cache") cache_path = os.path.join(cache_dir, f"{channel}-{post_id}-{file_unique_id}") if not os.path.exists(cache_path): logger.info(f"Background download started for {file_id}") await download_media_file(channel, post_id, file_unique_id) await asyncio.sleep(1) # Delay between downloads except Exception as e: logger.error(f"Background download failed for {file_id}: {str(e)}") continue await asyncio.sleep(60) # Check every minute except Exception as e: logger.error(f"Cache media files error: {str(e)}") await asyncio.sleep(60)