types: reduce MyPy debt without changing runtime behavior

.github/workflows/ci-cd.yml

@@ -56,8 +56,7 @@ jobs:
           poetry config virtualenvs.create false
           poetry install --with dev
 
-      - name: MyPy (known typing debt)
-        continue-on-error: true
+      - name: MyPy
         run: poetry run mypy app/ integrations/
 
   frontend:

app/core/logging.py

@@ -3,15 +3,25 @@
 import logging
 import sys
 from pathlib import Path
+from typing import Any, cast
 
 import structlog
 
-try:
-    import sentry_sdk
-    from sentry_sdk.integrations.aiohttp import AioHttpIntegration
-    from sentry_sdk.integrations.httpx import HttpxIntegration
-    from sentry_sdk.integrations.logging import LoggingIntegration
+sentry_sdk: Any
+LoggingIntegration: Any
+AioHttpIntegration: Any
+HttpxIntegration: Any
 
+try:
+    import sentry_sdk as _sentry_sdk
+    from sentry_sdk.integrations.aiohttp import AioHttpIntegration as _AioHttpIntegration
+    from sentry_sdk.integrations.httpx import HttpxIntegration as _HttpxIntegration
+    from sentry_sdk.integrations.logging import LoggingIntegration as _LoggingIntegration
+
+    sentry_sdk = _sentry_sdk
+    LoggingIntegration = _LoggingIntegration
+    AioHttpIntegration = _AioHttpIntegration
+    HttpxIntegration = _HttpxIntegration
     SENTRY_AVAILABLE = True
 except ImportError:  # pragma: no cover - optional dependency path
     sentry_sdk = None
@@ -35,7 +45,7 @@ def configure_logging(
     if debug:
         level = logging.DEBUG
 
-    shared_processors = [
+    shared_processors: list[structlog.types.Processor] = [
         structlog.contextvars.merge_contextvars,
         structlog.stdlib.add_log_level,
         structlog.stdlib.add_logger_name,
@@ -111,4 +121,4 @@ def capture_exception(exc: BaseException) -> None:
 
 
 def get_logger(name: str) -> structlog.stdlib.BoundLogger:
-    return structlog.get_logger(name)
+    return cast(structlog.stdlib.BoundLogger, structlog.get_logger(name))

app/scraper_worker.py

@@ -67,11 +67,11 @@ async def send_to_integrations(listings: list[ScrapedListing]) -> None:
     bitrix24_integration = get_bitrix24_integration()
     if bitrix24_integration is not None:
         try:
-            successful_ids = await bitrix24_integration.send_batch_deals(listings)
+            successful_deal_ids = await bitrix24_integration.send_batch_deals(listings)
             logger.info(
                 "Bitrix24 batch send completed",
                 count=len(listings),
-                success_count=len(successful_ids),
+                success_count=len(successful_deal_ids),
             )
         except Exception as exc:  # noqa: BLE001
             logger.warning("Bitrix24 integration failed", error=str(exc))

app/services/async_scraper.py

@@ -5,7 +5,7 @@
 import re
 from dataclasses import dataclass
 from datetime import date
-from typing import Sequence
+from typing import Any, Callable, Sequence
 from urllib.parse import parse_qs, urljoin, urlparse
 
 import httpx
@@ -14,13 +14,23 @@
 from app.core.config import Settings, SiteConfig
 from app.core.logging import get_logger
 
+extract_names: Callable[[str], list[str]] | None
+extract_phone_numbers: Callable[[str], list[str]] | None
+looks_like_person_name: Callable[[str], bool] | None
+normalize_phone_number: Callable[[str], str] | None
+
 try:
     from utils import (
-        extract_names,
-        extract_phone_numbers,
-        looks_like_person_name,
-        normalize_phone_number,
+        extract_names as _extract_names,
+        extract_phone_numbers as _extract_phone_numbers,
+        looks_like_person_name as _looks_like_person_name,
+        normalize_phone_number as _normalize_phone_number,
     )
+
+    extract_names = _extract_names
+    extract_phone_numbers = _extract_phone_numbers
+    looks_like_person_name = _looks_like_person_name
+    normalize_phone_number = _normalize_phone_number
 except ImportError:  # pragma: no cover - fallback path for isolated runtimes
     extract_names = None
     extract_phone_numbers = None
@@ -114,7 +124,7 @@ def __init__(self, site_config: SiteConfig, settings: Settings) -> None:
         )
         self._proxy_index = 0
 
-    def _get_client_kwargs(self) -> dict[str, object]:
+    def _get_client_kwargs(self) -> dict[str, Any]:
         headers = {
             "User-Agent": self._pick_user_agent(),
             "Accept": (
@@ -162,6 +172,17 @@ def _pick_proxy(self) -> str | None:
             return proxy
         return random.choice(pool)
 
+    @staticmethod
+    def _attr_str(tag: Tag | None, name: str, default: str = "") -> str:
+        if tag is None:
+            return default
+        value = tag.get(name)
+        if value is None:
+            return default
+        if isinstance(value, list):
+            return " ".join(str(item) for item in value)
+        return str(value)
+
     async def scrape(self) -> list[ScrapedListing]:
         today = date.today().isoformat()
         async with httpx.AsyncClient(**self._get_client_kwargs()) as client:
@@ -175,7 +196,7 @@ async def scrape(self) -> list[ScrapedListing]:
             listings: list[ScrapedListing] = []
             seen_ids: set[str] = set()
             for result in page_results:
-                if isinstance(result, Exception):
+                if isinstance(result, BaseException):
                     logger.exception(
                         "Page scraping failed",
                         site=self.site_config.name,
@@ -298,24 +319,24 @@ def _parse_imoti_listing_page(self, html: str, base_url: str) -> list[ScrapedLis
         soup = BeautifulSoup(html, "html.parser")
         exact_cards = soup.select("article.product-classic")
         if exact_cards:
-            results: list[ScrapedListing] = []
+            exact_results: list[ScrapedListing] = []
             seen_ids: set[str] = set()
             for article in exact_cards:
                 listing = self._parse_imoti_card_exact(article, base_url)
                 if listing is None or listing.ad_id in seen_ids:
                     continue
                 seen_ids.add(listing.ad_id)
                 if self._passes_filters(listing):
-                    results.append(listing)
-            return results
+                    exact_results.append(listing)
+            return exact_results
 
         soup = BeautifulSoup(html, "lxml")
         links = soup.select(self.site_config.selectors.get("link", "a[href*='/наеми/']"))
         results: list[ScrapedListing] = []
         seen_links: set[str] = set()
 
         for link_el in links:
-            link = self._normalize_link(base_url, link_el.get("href"))
+            link = self._normalize_link(base_url, self._attr_str(link_el, "href"))
             if not link or link in seen_links:
                 continue
             seen_links.add(link)
@@ -347,7 +368,7 @@ def _parse_imoti_card_exact(self, article: Tag, base_url: str) -> ScrapedListing
         if title_anchor is None:
             return None
 
-        link = self._normalize_link(base_url, title_anchor.get("href"))
+        link = self._normalize_link(base_url, self._attr_str(title_anchor, "href"))
         if not link:
             return None
 
@@ -416,7 +437,7 @@ def _parse_alo_card(self, card: Tag, base_url: str) -> ScrapedListing | None:
         if link_el is None:
             return None
 
-        link = self._normalize_link(base_url, link_el.get("href"))
+        link = self._normalize_link(base_url, self._attr_str(link_el, "href"))
         if not link:
             return None
 
@@ -430,7 +451,7 @@ def _parse_alo_card(self, card: Tag, base_url: str) -> ScrapedListing | None:
             title = self._title_from_url(link)
 
         image = card.select_one("a.avn_image img[alt], img[alt]")
-        image_alt = self._clean_text(image.get("alt", "") if image is not None else "")
+        image_alt = self._clean_text(self._attr_str(image, "alt"))
         card_text = self._clean_text(card.get_text(" ", strip=True))
         combined_text = self._clean_text(f"{title} {image_alt} {card_text}")
 
@@ -447,9 +468,7 @@ def _parse_alo_card(self, card: Tag, base_url: str) -> ScrapedListing | None:
 
         ad_id_match = re.search(
             r"adrows_(\d{4,12})",
-            " ".join(
-                card.get("id", []) if isinstance(card.get("id"), list) else [card.get("id", "")]
-            ),
+            self._attr_str(card, "id"),
         )
         ad_id = ad_id_match.group(1) if ad_id_match else self._extract_ad_id(link)
 
@@ -486,7 +505,7 @@ def _parse_generic_anchor_page(self, html: str, base_url: str) -> list[ScrapedLi
         seen_links: set[str] = set()
 
         for anchor in soup.select("a[href]"):
-            link = self._normalize_link(base_url, anchor.get("href"))
+            link = self._normalize_link(base_url, self._attr_str(anchor, "href"))
             if not link or link in seen_links or not self._link_looks_like_listing(link):
                 continue
 
@@ -518,7 +537,9 @@ def _parse_card(self, card: Tag, base_url: str) -> ScrapedListing | None:
         try:
             link_selector = self.site_config.selectors.get("link", "a[href]")
             link_el = card.select_one(link_selector)
-            link = self._normalize_link(base_url, link_el.get("href") if link_el else None)
+            link = self._normalize_link(
+                base_url, self._attr_str(link_el, "href") if link_el else None
+            )
             if not link:
                 return None
 
@@ -596,7 +617,7 @@ def _enrich_generic_detail(self, soup: BeautifulSoup, listing: ScrapedListing) -
 
         mailto_links = soup.select("a[href^='mailto:']")
         emails = [
-            a.get("href", "").split(":", 1)[-1].split("?", 1)[0].strip()
+            self._attr_str(a, "href").split(":", 1)[-1].split("?", 1)[0].strip()
             for a in mailto_links
             if a.get("href")
         ]
@@ -618,7 +639,7 @@ def _enrich_imoti_detail(self, soup: BeautifulSoup, listing: ScrapedListing) ->
 
         for block in soup.select("div.block-person-link"):
             icon = block.select_one("span.icon")
-            icon_classes = " ".join(icon.get("class", [])) if icon is not None else ""
+            icon_classes = self._attr_str(icon, "class")
             block_text = self._clean_text(block.get_text(" ", strip=True))
 
             if "mdi-account" in icon_classes and block_text:
@@ -629,7 +650,7 @@ def _enrich_imoti_detail(self, soup: BeautifulSoup, listing: ScrapedListing) ->
 
             if "mdi-phone" in icon_classes and not listing.phone:
                 tel = block.select_one("a[href^='tel:']")
-                phone_source = tel.get("href", "") if tel is not None else block_text
+                phone_source = self._attr_str(tel, "href") if tel is not None else block_text
                 phone = self._extract_phone_from_text(phone_source)
                 if phone:
                     listing.phone = phone
@@ -805,7 +826,7 @@ def _extract_contact_email_from_detail_soup(self, soup: BeautifulSoup) -> str:
         for root in soup.select("div.block-person-link, div.block-info, div.block-agent"):
             mail = root.select_one("a[href^='mailto:']")
             if mail is not None:
-                email = mail.get("href", "").replace("mailto:", "").split("?", 1)[0].strip()
+                email = self._attr_str(mail, "href").replace("mailto:", "").split("?", 1)[0].strip()
                 if email:
                     return email
             match = EMAIL_RE.search(root.get_text(" ", strip=True))
@@ -814,7 +835,7 @@ def _extract_contact_email_from_detail_soup(self, soup: BeautifulSoup) -> str:
 
         mail = soup.select_one("a[href^='mailto:']")
         if mail is not None:
-            email = mail.get("href", "").replace("mailto:", "").split("?", 1)[0].strip()
+            email = self._attr_str(mail, "href").replace("mailto:", "").split("?", 1)[0].strip()
             if email:
                 return email
 
@@ -875,7 +896,7 @@ def _extract_phone_from_text(self, text: str) -> str:
     def _extract_phone_from_detail_soup(self, soup: BeautifulSoup) -> str:
         tel_link = soup.select_one("a[href^='tel:']")
         if tel_link is not None:
-            phone = self._extract_phone_from_text(tel_link.get("href", ""))
+            phone = self._extract_phone_from_text(self._attr_str(tel_link, "href"))
             if phone:
                 return phone
 
@@ -976,7 +997,9 @@ def _extract_image(self, card: Tag | None, base_url: str) -> str:
         image = card.select_one("img[src], img[data-src]")
         if image is None:
             return ""
-        return self._normalize_link(base_url, image.get("src") or image.get("data-src"))
+        return self._normalize_link(
+            base_url, self._attr_str(image, "src") or self._attr_str(image, "data-src")
+        )
 
     def _extract_seller_name(self, card: Tag | None) -> str:
         if card is None:
@@ -1003,7 +1026,7 @@ def _guess_contact_name(self, text: str) -> str:
     def _extract_phone_from_imoti_card(self, article: Tag) -> str:
         tel_link = article.select_one('a[href^="tel:"]')
         if tel_link:
-            phone = self._extract_phone_from_text(tel_link.get("href", ""))
+            phone = self._extract_phone_from_text(self._attr_str(tel_link, "href"))
             if phone:
                 return phone
 
@@ -1079,7 +1102,7 @@ async def scrape_all_sites(self) -> list[ScrapedListing]:
 
         all_listings: list[ScrapedListing] = []
         for index, result in enumerate(results):
-            if isinstance(result, Exception):
+            if isinstance(result, BaseException):
                 logger.exception(
                     "Site scrape failed",
                     site=self.scrapers[index].site_config.name,

app/ui/streamlit_app.py

@@ -5,7 +5,7 @@
 import sys
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 import pandas as pd
 import streamlit as st
@@ -36,7 +36,7 @@
 def load_auth_config() -> dict[str, Any]:
     with open(settings.streamlit_users_yaml_path, "r", encoding="utf-8") as file:
         config = yaml.load(file, Loader=SafeLoader)
-    return config
+    return cast(dict[str, Any], config)
 
 
 def render_login() -> tuple[bool, str | None]:

app/voice/media.py

@@ -6,12 +6,15 @@
 import os
 import queue
 import threading
-from typing import Any
+from typing import Any, cast
 
 from app.core.logging import get_logger
 
+speech: Any
 try:
-    from google.cloud import speech
+    from google.cloud import speech as _speech
+
+    speech = _speech
 except ImportError:  # pragma: no cover - optional dependency path
     speech = None
 
@@ -104,7 +107,9 @@ def request_generator() -> Any:
                         break
                     yield speech.StreamingRecognizeRequest(audio_content=chunk)
 
-            responses = client.streaming_recognize(streaming_config, request_generator())
+            responses = client.streaming_recognize(
+                config=streaming_config, requests=request_generator()
+            )
             for response in responses:
                 for result in response.results:
                     if not result.alternatives:
@@ -139,7 +144,7 @@ async def handle_media_message(
         )
         manager.register_stream(call_sid, stream_sid)
         manager.ensure_stream(call_sid)
-        return call_sid
+        return cast(str, call_sid)
 
     if event == "media":
         media_payload = payload.get("media", {})
@@ -154,7 +159,7 @@ async def handle_media_message(
         encoded_audio = media_payload.get("payload", "")
         if encoded_audio:
             manager.push_audio(call_sid, base64.b64decode(encoded_audio))
-        return call_sid
+        return cast(str, call_sid)
 
     if event == "stop":
         stop_payload = payload.get("stop", {})
@@ -163,7 +168,7 @@ async def handle_media_message(
         )
         if call_sid:
             manager.stop_stream(call_sid)
-        return call_sid
+        return cast(str | None, call_sid)
 
     await asyncio.sleep(0)
     return None

app/voice/prompts.py

@@ -6,8 +6,11 @@
 
 from app.voice.extractors import QUESTION_SEQUENCE
 
+texttospeech: Any
 try:
-    from google.cloud import texttospeech
+    from google.cloud import texttospeech as _texttospeech
+
+    texttospeech = _texttospeech
 except ImportError:  # pragma: no cover - optional dependency path
     texttospeech = None