feat(telegram_bot): added api client for adnova

solution/services/telegram_bot/api/client.py

@@ -0,0 +1,168 @@
+from http import HTTPStatus as status
+from typing import Self
+
+import httpx
+
+import config
+from api import errors, schemas
+
+
+class AdNovaClient:
+    def __init__(self) -> None:
+        self.base_url = config.API_ENDPOINT
+
+    async def __aenter__(self) -> Self:
+        self.client = httpx.AsyncClient(base_url=self.base_url)
+        return self
+
+    async def __aexit__(
+        self, exc_type: object, exc_val: object, exc_tb: object
+    ) -> None:
+        await self.client.aclose()
+
+    def _handle_response(self, response: httpx.Response) -> httpx.Response:
+        if response.status_code == status.BAD_REQUEST:
+            error = schemas.BadRequestError.model_validate(response.json())
+            raise errors.BadRequestError(error.detail)
+        if response.status_code == status.FORBIDDEN:
+            error = schemas.ForbiddenError.model_validate(response.json())
+            raise errors.ForbiddenError(error.detail)
+        if response.status_code == status.NOT_FOUND:
+            error = schemas.NotFoundError.model_validate(response.json())
+            raise errors.NotFoundError(error.detail)
+
+        response.raise_for_status()
+
+        return response
+
+    def sync_get_advertiser(self, advertiser_id: str) -> schemas.Advertiser:
+        client = httpx.Client(base_url=self.base_url)
+        response = client.get(f"/advertisers/{advertiser_id}")
+        self._handle_response(response)
+        return schemas.Advertiser.model_validate(response.json())
+
+    async def get_advertiser(self, advertiser_id: str) -> schemas.Advertiser:
+        response = await self.client.get(f"/advertisers/{advertiser_id}")
+        self._handle_response(response)
+        return schemas.Advertiser.model_validate(response.json())
+
+    async def create_campaign(
+        self, advertiser_id: str, data: schemas.CampaignCreateIn
+    ) -> schemas.CampaignOut:
+        response = await self.client.post(
+            f"/advertisers/{advertiser_id}/campaigns", json=data.model_dump()
+        )
+        self._handle_response(response)
+        return schemas.CampaignOut.model_validate(response.json())
+
+    async def list_campaigns(
+        self, advertiser_id: str, page: int = 1, size: int = 100
+    ) -> list[schemas.CampaignOut]:
+        params = {"page": page, "size": size}
+        response = await self.client.get(
+            f"/advertisers/{advertiser_id}/campaigns", params=params
+        )
+        self._handle_response(response)
+        return [
+            schemas.CampaignOut.model_validate(item)
+            for item in response.json()
+        ]
+
+    async def get_campaign(
+        self, advertiser_id: str, campaign_id: str
+    ) -> schemas.CampaignOut:
+        response = await self.client.get(
+            f"/advertisers/{advertiser_id}/campaigns/{campaign_id}"
+        )
+        self._handle_response(response)
+        return schemas.CampaignOut.model_validate(response.json())
+
+    async def update_campaign(
+        self,
+        advertiser_id: str,
+        campaign_id: str,
+        data: schemas.CampaignUpdateIn,
+    ) -> schemas.CampaignOut:
+        response = await self.client.put(
+            f"/advertisers/{advertiser_id}/campaigns/{campaign_id}",
+            json=data.model_dump(),
+        )
+        self._handle_response(response)
+        return schemas.CampaignOut.model_validate(response.json())
+
+    async def delete_campaign(
+        self, advertiser_id: str, campaign_id: str
+    ) -> None:
+        response = await self.client.delete(
+            f"/advertisers/{advertiser_id}/campaigns/{campaign_id}"
+        )
+        self._handle_response(response)
+
+    async def upload_ad_image(
+        self, advertiser_id: str, campaign_id: str, file: bytes
+    ) -> schemas.CampaignOut:
+        files = {"ad_image": file}
+        response = await self.client.post(
+            f"/advertisers/{advertiser_id}/campaigns/{campaign_id}/ad_image/upload",
+            files=files,
+        )
+        self._handle_response(response)
+        return schemas.CampaignOut.model_validate(response.json())
+
+    async def delete_ad_image(
+        self, advertiser_id: str, campaign_id: str
+    ) -> None:
+        response = await self.client.delete(
+            f"/advertisers/{advertiser_id}/campaigns/{campaign_id}/ad_image/delete"
+        )
+        self._handle_response(response)
+
+    async def get_advertiser_statistics(
+        self, advertiser_id: str
+    ) -> schemas.Stat:
+        response = await self.client.get(f"/stats/advertisers/{advertiser_id}")
+        self._handle_response(response)
+        return schemas.Stat.model_validate(response.json())
+
+    async def get_daily_advertiser_statistics(
+        self, advertiser_id: str
+    ) -> list[schemas.DailyStat]:
+        response = await self.client.get(
+            f"/stats/advertisers/{advertiser_id}/daily"
+        )
+        self._handle_response(response)
+        return [
+            schemas.DailyStat.model_validate(item) for item in response.json()
+        ]
+
+    async def get_campaign_statistics(self, campaign_id: str) -> schemas.Stat:
+        response = await self.client.get(f"/stats/campaigns/{campaign_id}")
+        self._handle_response(response)
+        return schemas.Stat.model_validate(response.json())
+
+    async def get_daily_campaign_statistics(
+        self, campaign_id: str
+    ) -> list[schemas.DailyStat]:
+        response = await self.client.get(
+            f"/stats/campaigns/{campaign_id}/daily"
+        )
+        self._handle_response(response)
+        return [
+            schemas.DailyStat.model_validate(item) for item in response.json()
+        ]
+
+    async def generate_ad_text(
+        self, data: schemas.GenerateAdTextIn
+    ) -> schemas.GenerateAdTextResult:
+        response = await self.client.post(
+            "/generate/ad_text", json=data.model_dump()
+        )
+        self._handle_response(response)
+        return schemas.GenerateAdTextResult.model_validate(response.json())
+
+    async def get_generate_ad_text_result(
+        self, task_id: str
+    ) -> schemas.GenerateAdTextResult:
+        response = await self.client.get(f"/generate/ad_text/{task_id}/result")
+        self._handle_response(response)
+        return schemas.GenerateAdTextResult.model_validate(response.json())

solution/services/telegram_bot/api/errors.py

@@ -0,0 +1,23 @@
+from typing import Any
+
+
+class HTTPError(Exception):
+    pass
+
+
+class BadRequestError(HTTPError):
+    def __init__(self, detail: Any) -> None:
+        super().__init__(f"Bad Request: {detail}")
+        self.detail = detail
+
+
+class ForbiddenError(HTTPError):
+    def __init__(self, detail: str = "Forbidden") -> None:
+        super().__init__(f"Forbidden: {detail}")
+        self.detail = detail
+
+
+class NotFoundError(HTTPError):
+    def __init__(self, detail: str = "Not Found") -> None:
+        super().__init__(f"Not Found: {detail}")
+        self.detail = detail

solution/services/telegram_bot/api/schemas.py

@@ -0,0 +1,85 @@
+from typing import Annotated, Any, Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field, NonNegativeFloat, NonNegativeInt
+
+
+class BadRequestError(BaseModel):
+    detail: Any
+
+
+class ForbiddenError(BaseModel):
+    detail: str = "Forbidden"
+
+
+class NotFoundError(BaseModel):
+    detail: str = "Not Found"
+
+
+class CampaignTargeting(BaseModel):
+    gender: Literal["MALE", "FEMALE", "ALL"] | None = None
+    age_from: Annotated[NonNegativeInt, Field(strict=True, ls=100)] | None = (
+        None
+    )
+    age_to: Annotated[NonNegativeInt, Field(strict=True, ls=100)] | None = None
+    location: str | None = None
+
+
+class CampaignCreateIn(BaseModel):
+    targeting: CampaignTargeting
+    ad_title: str
+    ad_text: str
+    impressions_limit: NonNegativeInt
+    clicks_limit: NonNegativeInt
+    cost_per_impression: NonNegativeFloat
+    cost_per_click: NonNegativeFloat
+    start_date: NonNegativeInt
+    end_date: NonNegativeInt
+
+
+class CampaignUpdateIn(CampaignCreateIn):
+    pass
+
+
+class CampaignOut(BaseModel):
+    campaign_id: str
+    advertiser_id: str
+    targeting: CampaignTargeting
+    ad_title: str
+    ad_text: str
+    ad_image: str | None = None
+    impressions_limit: NonNegativeInt
+    clicks_limit: NonNegativeInt
+    cost_per_impression: NonNegativeFloat
+    cost_per_click: NonNegativeFloat
+    start_date: NonNegativeInt
+    end_date: NonNegativeInt
+
+
+class Advertiser(BaseModel):
+    advertiser_id: UUID
+    name: str
+
+
+class Stat(BaseModel):
+    impressions_count: int
+    clicks_count: int
+    conversion: float
+    spent_impressions: float
+    spent_clicks: float
+    spent_total: float
+
+
+class DailyStat(Stat):
+    date: int
+
+
+class GenerateAdTextIn(BaseModel):
+    advertiser_name: str
+    ad_title: str
+
+
+class GenerateAdTextResult(BaseModel):
+    task_id: str
+    status: str
+    result: str