add types-2

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)

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:

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'<div class="document-pdf" style="padding: 10px;"><a href="{tg_link}" target="_blank">[PDF-файл]</a></div>')
+                    # 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'<div class="document-pdf" style="padding: 10px;"><a href="{tg_link}" target="_blank">[PDF-файл]</a></div>')
+                    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'<div class="document-pdf" style="padding: 10px;">[PDF-файл]</div>') # Add placeholder without link
                 elif message.media in [MessageMediaType.PHOTO, MessageMediaType.DOCUMENT]:
                     content_media.append(f'<img src="{url}" style="max-width:100%; width:auto; height:auto; max-height:400px; object-fit:contain;">')
                 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

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: