diff --git a/api_server.py b/api_server.py index ec375eb..ca2544c 100644 --- a/api_server.py +++ b/api_server.py @@ -12,7 +12,7 @@ import logging import os import mimetypes -from typing import List, Union +from typing import List, Union, Any import json from datetime import datetime @@ -28,6 +28,7 @@ import json_repair import magic from pyrogram import errors +from pyrogram.types import Message from fastapi import FastAPI, HTTPException, Response, Request from fastapi.responses import HTMLResponse, FileResponse from telegram_client import TelegramClient @@ -114,7 +115,7 @@ if __name__ == "__main__": loop="uvloop" ) -async def find_file_id_in_message(message, file_unique_id: str): +async def find_file_id_in_message(message: Message, file_unique_id: str) -> Union[str, None]: """Find file_id by checking all possible media types in message""" if message.media == "MessageMediaType.POLL": logger.debug(f"Message {message.id} is a poll, skipping media search") @@ -159,11 +160,12 @@ async def find_file_id_in_message(message, file_unique_id: str): return message.document.file_id # If we reached here, the file_unique_id was not found - logger.warning(f"Could not find media with file_unique_id '{file_unique_id}' in message {message.id} (channel: {message.chat.id}). Found media: {', '.join(media_found) or 'None'}") + channel_id_log = message.chat.id if message.chat else 'unknown_chat' + logger.warning(f"Could not find media with file_unique_id '{file_unique_id}' in message {message.id} (channel: {channel_id_log}). Found media: {', '.join(media_found) or 'None'}") return None -def delayed_delete_file(file_path: str, delay: int = 300): +def delayed_delete_file(file_path: str, delay: int = 300) -> None: """ Delete temporary file after a delay to ensure complete file delivery. Delay is set to 5 minutes by default. @@ -176,7 +178,7 @@ def delayed_delete_file(file_path: str, delay: int = 300): logger.error(f"Failed to delete temporary file {file_path}: {str(e)}") -async def prepare_file_response(file_path: str, delete_after: bool = False): +async def prepare_file_response(file_path: str, delete_after: bool = False) -> FileResponse: """Prepare file response with proper headers""" if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found") @@ -410,7 +412,7 @@ async def remove_old_cached_files(media_files: list, cache_dir: str) -> tuple[li return updated_media_files, files_removed -async def download_new_files(media_files: list, cache_dir: str): +async def download_new_files(media_files: list, cache_dir: str) -> None: """ Download files that are not in cache yet """ @@ -479,7 +481,7 @@ def fix_corrupted_json(file_path: str) -> list: logger.error(f"Failed to fix JSON file {file_path}: {str(e)}") return [] -async def cache_media_files(): +async def cache_media_files() -> None: """Background task for cache management: removes old files and downloads new ones""" delay = 60 while True: @@ -522,7 +524,7 @@ async def cache_media_files(): await asyncio.sleep(delay) -def calculate_cache_stats(): +def calculate_cache_stats() -> dict[str, Any]: """ Calculate cache statistics including file count, total size in MB, and time difference in days. Returns a dictionary with keys: 'cache_files_count', 'cache_total_size_mb', 'cache_time_diff_days', 'channels'. @@ -535,8 +537,8 @@ def calculate_cache_stats(): if os.path.isdir(base_cache_dir): # Recursively walk through all subdirectories for root, _, files in os.walk(base_cache_dir): - for f in files: - file_path = os.path.join(root, f) + for current_file in files: + file_path = os.path.join(root, current_file) file_size = os.path.getsize(file_path) cache_files_count += 1 cache_total_size_bytes += file_size @@ -548,7 +550,7 @@ def calculate_cache_stats(): if channel not in channels_stats: channels_stats[channel] = { 'files_count': 0, - 'size_mb': 0 + 'size_mb': 0.0 } channels_stats[channel]['files_count'] += 1 @@ -565,8 +567,8 @@ def calculate_cache_stats(): cache_times = [] if os.path.exists(media_file_ids_path): try: - with open(media_file_ids_path, "r", encoding="utf-8") as f: - media_files = json.load(f) + with open(media_file_ids_path, "r", encoding="utf-8") as json_file: + media_files = json.load(json_file) for entry in media_files: if "added" in entry: cache_times.append(entry["added"]) @@ -596,7 +598,7 @@ def is_local_request(request: Union[Request, None]) -> bool: @app.get("/post/html/{channel}/{post_id}", response_class=HTMLResponse) @app.get("/html/{channel}/{post_id}/{token}", response_class=HTMLResponse) @app.get("/post/html/{channel}/{post_id}/{token}", response_class=HTMLResponse) -async def get_post_html(channel: str, post_id: int, request: Request, token: str | None = None, debug: bool = False): +async def get_post_html(channel: str, post_id: int, request: Request, token: str | None = None, debug: bool = False) -> HTMLResponse: if Config["token"] and is_local_request(request): if token != Config["token"]: logger.error(f"Invalid token for HTML post: {token}, expected: {Config['token']}") @@ -609,7 +611,7 @@ async def get_post_html(channel: str, post_id: int, request: Request, token: str html_content = await parser.get_post(channel, post_id, 'html', debug) if not html_content: raise HTTPException(status_code=404, detail="Post not found") - return html_content + return HTMLResponse(content=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) @@ -620,7 +622,7 @@ async def get_post_html(channel: str, post_id: int, request: Request, token: str @app.get("/post/json/{channel}/{post_id}") @app.get("/json/{channel}/{post_id}/{token}") @app.get("/post/json/{channel}/{post_id}/{token}") -async def get_post(channel: str, post_id: int, request: Request, token: str | None = None, debug: bool = False): +async def get_post(channel: str, post_id: int, request: Request, token: str | None = None, debug: bool = False) -> Response: if Config["token"] and is_local_request(request): if token != Config["token"]: logger.error(f"Invalid token for JSON post: {token}, expected: {Config['token']}") @@ -633,7 +635,7 @@ async def get_post(channel: str, post_id: int, request: Request, token: str | No json_content = await parser.get_post(channel, post_id, 'json', debug) if not json_content: raise HTTPException(status_code=404, detail="Post not found") - return json_content + return Response(content=json_content, media_type="application/json") 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) @@ -642,7 +644,7 @@ async def get_post(channel: str, post_id: int, request: Request, token: str | No @app.get("/raw_json/{channel}/{post_id}") @app.get("/raw_json/{channel}/{post_id}/{token}") -async def get_raw_post_json(channel: str, post_id: int, request: Request, token: str | None = None): +async def get_raw_post_json(channel: str, post_id: int, request: Request, token: str | None = None) -> Response: if Config["token"] and is_local_request(request): if token != Config["token"]: logger.error(f"Invalid token for raw JSON post: {token}, expected: {Config['token']}") @@ -670,7 +672,7 @@ async def get_raw_post_json(channel: str, post_id: int, request: Request, token: @app.get("/health") @app.get("/health/{token}") -async def health_check(request: Request, token: str | None = None): +async def health_check(request: Request, token: str | None = None) -> Response: if Config["token"] and is_local_request(request): if token != Config["token"]: logger.error(f"Invalid token for health check: {token}, expected: {Config['token']}") @@ -690,7 +692,7 @@ async def health_check(request: Request, token: str | None = None): else: config_info[config_key] = config_value - return { + data = { "status": "ok", "tg_connected": client.client.is_connected, "tg_name": me.username, @@ -701,6 +703,7 @@ async def health_check(request: Request, token: str | None = None): "config": config_info, **cache_stats } + return Response(content=json.dumps(data), media_type="application/json") except Exception as e: error_message = f"Failed to get health check: {str(e)}" logger.error(error_message) @@ -708,7 +711,7 @@ async def health_check(request: Request, token: str | None = None): @app.get("/media/{channel}/{post_id}/{file_unique_id}/{digest}") @app.get("/media/{channel}/{post_id}/{file_unique_id}") -async def get_media(channel: str, post_id: int, file_unique_id: str, digest: str | None = None): +async def get_media(channel: str, post_id: int, file_unique_id: str, digest: str | None = None) -> FileResponse|Response: try: url = f"{channel}/{post_id}/{file_unique_id}" if not verify_media_digest(url, digest): @@ -751,6 +754,11 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, digest: str logger.error(error_message) raise HTTPException(status_code=500, detail=error_message) from e + # If the code reaches here, something is wrong with the logic above. + # This part should theoretically be unreachable. + logger.error(f"get_media reached unexpected state for {channel}/{post_id}/{file_unique_id}") + raise HTTPException(status_code=500, detail="Internal server error: Unexpected state reached in media handling.") + @app.get("/rss/{channel}", response_class=Response) @app.get("/rss/{channel}/{token}", response_class=Response) async def get_rss_feed(channel: str, @@ -761,7 +769,7 @@ async def get_rss_feed(channel: str, exclude_flags: str | None = None, exclude_text: str | None = None, merge_seconds: int = 5, - ): + ) -> Response: if Config["token"] and is_local_request(request): if token != Config["token"]: logger.error(f"invalid_token_error: token {token}, expected {Config['token']}") @@ -808,7 +816,7 @@ async def get_rss_feed(channel: str, @app.get("/flags", response_model=List[str]) @app.get("/flags/{token}", response_model=List[str]) -async def get_available_flags(request: Request, token: str | None = None): +async def get_available_flags(request: Request, token: str | None = None) -> Response: """Returns a list of all possible flags that can be assigned to posts.""" if Config["token"] and is_local_request(request): if token != Config["token"]: @@ -819,7 +827,7 @@ async def get_available_flags(request: Request, token: str | None = None): try: flags = PostParser.get_all_possible_flags() - return flags + return Response(content=flags, media_type="text/plain") except Exception as e: error_message = f"Failed to get flags list: {str(e)}" logger.error(error_message) diff --git a/config.py b/config.py index 3e79607..57ee445 100644 --- a/config.py +++ b/config.py @@ -9,8 +9,9 @@ # pylance: disable=reportMissingImports, reportMissingModuleSource import os +from typing import Any -def get_settings(): +def get_settings() -> dict[str, Any]: tg_api_id = os.getenv("TG_API_ID") tg_api_hash = os.getenv("TG_API_HASH") if not tg_api_id or not tg_api_hash: diff --git a/post_parser.py b/post_parser.py index 7486dbe..8ce075b 100644 --- a/post_parser.py +++ b/post_parser.py @@ -83,23 +83,26 @@ class PostParser: # but for now, we return an empty list. return [] - def channel_name_prepare(self, channel: str): + def channel_name_prepare(self, channel: str | int) -> Union[str, int]: if isinstance(channel, str) and channel.startswith('-100'): # Convert numeric channel ID to int channel_id = int(channel) return channel_id else: return channel - async def get_post(self, channel: str, post_id: int, output_type: str = 'json', debug: bool = False) -> Union[str, Dict[Any, Any], None]: + async def get_post(self, channel: str, + post_id: int, + output_type: str = 'json', + debug: bool = False) -> Union[str, Dict[Any, Any], None]: print(f"Getting post {channel}, {post_id}") try: - channel = self.channel_name_prepare(channel) - message = await self.client.get_messages(channel, post_id) + prepared_channel_id: Union[str, int] = self.channel_name_prepare(channel) + message = await self.client.get_messages(prepared_channel_id, post_id) if Config["debug"]: print(message) if not message or getattr(message, 'empty', False): - logger.error(f"post_not_found_or_empty: channel {channel}, post_id {post_id}") + logger.error(f"post_not_found_or_empty: channel {prepared_channel_id}, post_id {post_id}") return None if output_type == 'html': @@ -111,7 +114,7 @@ class PostParser: return None except Exception as e: - logger.error(f"post_parsing_error: channel {channel}, post_id {post_id}, error {str(e)}") + logger.error(f"post_parsing_error: channel {prepared_channel_id}, post_id {post_id}, error {str(e)}") raise def _get_author_info(self, message: Message) -> str: #Tests: tests/postparser_author_info.py @@ -588,9 +591,15 @@ class PostParser: # Check if document is a PDF file if message.media == MessageMediaType.DOCUMENT and message.document is not None and hasattr(message.document, 'mime_type') and message.document.mime_type == 'application/pdf': - if channel_username.startswith('-100'): tg_link = f"https://t.me/c/{channel_username[4:]}/{message.id}" - else: tg_link = f"https://t.me/{channel_username}/{message.id}" - content_media.append(f'
[PDF-файл]
') + # Only attempt to create link if channel_username is available + if channel_username: + if channel_username.startswith('-100'): tg_link = f"https://t.me/c/{channel_username[4:]}/{message.id}" + else: tg_link = f"https://t.me/{channel_username}/{message.id}" + content_media.append(f'
[PDF-файл]
') + else: + # Handle case where channel_username is None (e.g., log or add placeholder) + logger.warning(f"Could not generate PDF link for message {message.id} because channel username is missing.") + content_media.append(f'
[PDF-файл]
') # Add placeholder without link elif message.media in [MessageMediaType.PHOTO, MessageMediaType.DOCUMENT]: content_media.append(f'') elif message.media == MessageMediaType.VIDEO: @@ -894,7 +903,7 @@ class PostParser: except Exception as e: logger.error(f"file_id_collection_error: message_id {message.id}, error {str(e)}") - def get_channel_username(self, message): + def get_channel_username(self, message: Message) -> Union[str, None]: """Extract channel username or ID from message""" chat = message.chat if hasattr(message, 'chat') else message if not chat: return None diff --git a/rss_generator.py b/rss_generator.py index b01e972..0b4ae68 100644 --- a/rss_generator.py +++ b/rss_generator.py @@ -14,7 +14,7 @@ import re from datetime import datetime, timezone from typing import Optional from feedgen.feed import FeedGenerator -from pyrogram import errors +from pyrogram import errors, Client from pyrogram.types import Message from post_parser import PostParser from config import get_settings @@ -30,24 +30,24 @@ if not logger.handlers: logger.addHandler(handler) -async def _create_time_based_media_groups(messages, merge_seconds: int = 5): +async def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]: """ Create media groups based on time difference between messages """ - messages_sorted = sorted(messages, key=lambda x: x.date) + messages_sorted = sorted(messages, key=lambda msg: msg.date) # type: ignore cluster: list[Message] = [] - last_msg_date: Optional[datetime] = None + last_msg_date: datetime = datetime.now(timezone.utc) current_media_group_id: Optional[str] = None for msg in messages_sorted: if not cluster: cluster.append(msg) - last_msg_date = msg.date + last_msg_date = msg.date # type: ignore current_media_group_id = getattr(msg, "media_group_id", None) continue - time_diff = (msg.date - last_msg_date).total_seconds() + time_diff = (msg.date - last_msg_date).total_seconds() # type: ignore msg_media_group_id = getattr(msg, "media_group_id", None) @@ -59,7 +59,7 @@ async def _create_time_based_media_groups(messages, merge_seconds: int = 5): for m in cluster: m.media_group_id = current_media_group_id # type: ignore cluster.append(msg) - last_msg_date = msg.date + last_msg_date = msg.date # type: ignore else: if len(cluster) >= 2 and not current_media_group_id: dates = [m.date for m in cluster if m.date is not None] @@ -69,7 +69,7 @@ async def _create_time_based_media_groups(messages, merge_seconds: int = 5): for m in cluster: m.media_group_id = new_group_id # type: ignore cluster = [msg] - last_msg_date = msg.date + last_msg_date = msg.date # type: ignore current_media_group_id = msg_media_group_id if len(cluster) >= 2 and not current_media_group_id: @@ -82,12 +82,12 @@ async def _create_time_based_media_groups(messages, merge_seconds: int = 5): return messages_sorted -async def _create_messages_groups(messages): +async def _create_messages_groups(messages: list[Message]) -> list[list[Message]]: """ Process messages into formatted posts, handling media groups """ - processing_groups = [] - media_groups = {} + processing_groups: list[list[Message]] = [] + media_groups: dict[str | int, list[Message]] = {} # First pass - collect messages and organize into processing groups for message in messages: @@ -113,7 +113,8 @@ async def _create_messages_groups(messages): processing_groups.append([message]) # Single message becomes its own processing group except Exception as e: - logger.error(f"_create_messages_groups: channel {message.chat.username}, message_id {message.id}, error {str(e)}") + username = message.chat.username if message.chat else 'unknown_chat' + logger.error(f"_create_messages_groups: channel {username}, message_id {message.id}, error {str(e)}") continue # Sort messages within media groups by message ID in descending order @@ -122,11 +123,11 @@ async def _create_messages_groups(messages): processing_groups.append(media_group) # Sort processing groups by date of first message in each group - processing_groups.sort( key=lambda group: group[0].date, reverse=True ) + processing_groups.sort(key=lambda group: group[0].date if group[0].date else datetime.now(timezone.utc), reverse=True) return processing_groups -async def _trim_messages_groups(messages_groups, limit): +async def _trim_messages_groups(messages_groups: list[list[Message]], limit: int): """ Trim messages groups to limit """ @@ -138,7 +139,10 @@ async def _trim_messages_groups(messages_groups, limit): return messages_groups -async def _render_messages_groups(messages_groups, post_parser, exclude_flags: str | None = None, exclude_text: str | None = None): +async def _render_messages_groups(messages_groups: list[list[Message]], + post_parser: PostParser, + exclude_flags: str | None = None, + exclude_text: str | None = None): """ Render message groups into HTML format Args: @@ -193,7 +197,7 @@ async def _render_messages_groups(messages_groups, post_parser, exclude_flags: s for msg in processed_messages: if msg.get('flags'): # Check if flags exist and are not empty all_flags.update(msg['flags']) - all_flags.update("merged") + all_flags.add("merged") merged_flags = list(all_flags) # Convert back to list if needed, or keep as set footer_html = post_parser.generate_html_footer(main_message, flags_list=merged_flags) @@ -250,9 +254,8 @@ async def _render_messages_groups(messages_groups, post_parser, exclude_flags: s rendered_posts.sort(key=lambda x: x['date'], reverse=True) return rendered_posts -async def generate_channel_rss(channel: str, - post_parser: Optional[PostParser] = None, #TODO избавиться от post_parser в аргументах - client = None, +async def generate_channel_rss(channel: str | int, + client: Client, limit: int = 20, exclude_flags: str | None = None, exclude_text: str | None = None, @@ -276,16 +279,14 @@ async def generate_channel_rss(channel: str, raise ValueError(f"limit cannot exceed 200, got {limit}") try: - if post_parser is None: - post_parser = PostParser(client=client) + post_parser = PostParser(client=client) fg = FeedGenerator() fg.load_extension('dc') base_url = Config['pyrogram_bridge_url'] try: - channel = post_parser.channel_name_prepare(channel) - logger.debug(f"Prepared channel identifier: {channel} (type: {type(channel)})") # Log prepared channel + channel = post_parser.channel_name_prepare(channel) channel_info = await post_parser.client.get_chat(channel) channel_title = channel_info.title or f"Telegram: {channel}" channel_username = post_parser.get_channel_username(channel_info) @@ -354,9 +355,8 @@ async def generate_channel_rss(channel: str, raise -async def generate_channel_html(channel: str, - post_parser: Optional[PostParser] = None, - client = None, +async def generate_channel_html(channel: str | int, + client: Client, limit: int = 20, exclude_flags: str | None = None, exclude_text: str | None = None, @@ -380,8 +380,7 @@ async def generate_channel_html(channel: str, raise ValueError(f"limit cannot exceed 200, got {limit}") try: - if post_parser is None: - post_parser = PostParser(client=client) + post_parser = PostParser(client=client) base_url = Config['pyrogram_bridge_url'] @@ -427,7 +426,7 @@ async def generate_channel_html(channel: str, logger.error(f"html_generation_error: channel {channel}, error {str(e)}") raise -def create_error_feed(channel: str, base_url: str) -> str: +def create_error_feed(channel: str | int, base_url: str) -> str: """ Create an empty RSS feed with metadata indicating an error when the channel is not found. Args: