PROD.2025/AdNova
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(telegram_bot): added campaign creation and updating, added help command

Browse Source
This commit is contained in:
ITQ
2025-02-22 07:26:04 +03:00
parent ec47e7754e
commit c0f35512a1
10 changed files with 349 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
solution/services/telegram_bot/api/schemas.py
+4 -2
View File
@@ -18,10 +18,12 @@ class NotFoundError(BaseModel):
class CampaignTargeting(BaseModel):
gender: Literal["MALE", "FEMALE", "ALL"] | None = None
age_from: Annotated[NonNegativeInt, Field(strict=True, ls=100)] | None = (
age_from: Annotated[NonNegativeInt, Field(strict=False, ls=100)] | None = (
None
)
age_to: Annotated[NonNegativeInt, Field(strict=False, ls=100)] | None = (
None
)
age_to: Annotated[NonNegativeInt, Field(strict=True, ls=100)] | None = None
location: str | None = None
solution/services/telegram_bot/commands/help.py
+21
View File
@@ -0,0 +1,21 @@
from aiogram import Router
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.types import Message
from aiogram_dialog import DialogManager
help_router = Router()
@help_router.message(Command("help"))
async def help_command(
message: Message, dialog_manager: DialogManager, state: FSMContext
) -> None:
response = (
"Commands:\n\n"
"/start - Start the bot and authenticate as advertiser\n"
"/campaigns - Manage your campaigns\n"
"/statistics - See your overall statistics\n"
"/logout - Logout of current advertiser account"
)
await message.answer(response)
solution/services/telegram_bot/commands/start.py
+2 -1
View File
@@ -19,7 +19,8 @@ async def start_command(
await message.answer(
"Already authenticated as"
f" <code>{state_data['advertiser']['name']}</code> "
f"(<code>{state_data['advertiser']['advertiser_id']}</code>)"
f"(<code>{state_data['advertiser']['advertiser_id']}</code>)."
"Get all commands with /help."
)
return
solution/services/telegram_bot/commands/stats.py
+3 -3
View File
@@ -26,8 +26,8 @@ async def stats_command(
f"\t• Impressions: {stats.impressions_count}\n"
f"\t• Clicks: {stats.clicks_count}\n"
f"\t• Conversion: {stats.conversion:.2f}%\n"
f"\t• Spent on impressions: ${stats.spent_impressions:.2f}\n"
f"\t• Spent on clicks: ${stats.spent_clicks:.2f}\n"
f"\t• Spent total: ${stats.spent_total:.2f}"
f"\t• Spent on impressions: {stats.spent_impressions:.2f}\n"
f"\t• Spent on clicks: {stats.spent_clicks:.2f}\n"
f"\t• Spent total: {stats.spent_total:.2f}"
)
await message.answer(response)
solution/services/telegram_bot/dialogs/campaigns.py
+292 -48
View File
@@ -6,18 +6,72 @@ from urllib.parse import urlparse
import httpx
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery, ContentType
from aiogram.types import CallbackQuery, ContentType, Message
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.api.entities import MediaAttachment
from aiogram_dialog.widgets.common import Whenable
from aiogram_dialog.widgets.input import ManagedTextInput, TextInput
from aiogram_dialog.widgets.kbd import Button, ListGroup, ScrollingGroup, Start
from aiogram_dialog.widgets.media import DynamicMedia
from aiogram_dialog.widgets.text import Const, Format
from pydantic import ValidationError
import config
from api.client import AdNovaClient
from api.errors import BadRequestError, ForbiddenError
from dialogs import utils
from states.campaigns import CampaignsDailogState
campaign_info = (
Const('<pre><code class="language-input-format">Title\n'),
Const("Text\n"),
Const("Impressions limit\n"),
Const("Clicks limit\n"),
Const("Cost per impression\n"),
Const("Cost per click\n"),
Const("Start date\n"),
Const("End date\n"),
Const("Targeting gender (ALL, FEMALE, MALE, None)\n"),
Const("Targeting age from (could be None)\n"),
Const("Targeting age to (could be None)\n"),
Const("Targeting location (could be None)"),
Const("</code></pre>"),
Const('<pre><code class="language-example">Some title\n'),
Const("Some text\n"),
Const("15\n"),
Const("10\n"),
Const("0.4\n"),
Const("0.5\n"),
Const("100\n"),
Const("110\n"),
Const("ALL\n"),
Const("12\n"),
Const("None\n"),
Const("Moscow"),
Const("</code></pre>"),
)
def campaign_has_ad_image(
data: dict, widget: Whenable, manager: DialogManager
) -> bool:
return bool(data["dialog_data"]["campaign"]["ad_image"])
def campaign_has_not_ad_image(
data: dict, widget: Whenable, manager: DialogManager
) -> bool:
return not data["dialog_data"]["campaign"]["ad_image"]
def check_campaign(campaign_data: str) -> None:
fields = campaign_data.split("\n\n")
if len(fields) != 12:
raise ValueError
utils.campaign_from_list(fields)
async def campaigns(**kwargs: dict[Any]) -> dict[str, Any]:
state: FSMContext = kwargs["state"]
@@ -37,50 +91,10 @@ async def campaigns(**kwargs: dict[Any]) -> dict[str, Any]:
}
async def campaign_detail_on_click(
callback: CallbackQuery, button: Button, manager: DialogManager
) -> None:
manager_data: dict[Any] = await manager.load_data()
state: FSMContext = manager_data["middleware_data"]["state"]
state_data = await state.get_data()
advertiser_id = state_data["advertiser_id"]
campaign_id = manager.item_id
if campaign_id == "‎":
return
async with AdNovaClient() as client:
campaign = await client.get_campaign(
advertiser_id=advertiser_id,
campaign_id=campaign_id,
)
if campaign.ad_image:
campaign.ad_image = (
f"{config.MINIO_URL}{urlparse(campaign.ad_image).path}"
)
await manager.update({"campaign": campaign.model_dump(mode="json")})
await callback.answer()
await manager.switch_to(CampaignsDailogState.campaign)
def campaign_has_ad_image(
data: dict, widget: Whenable, manager: DialogManager
) -> bool:
return bool(data["dialog_data"]["campaign"]["ad_image"])
def campaign_has_not_ad_image(
data: dict, widget: Whenable, manager: DialogManager
) -> bool:
return not data["dialog_data"]["campaign"]["ad_image"]
async def campaign_by_id(**kwargs: dict[Any]) -> dict[str, Any]:
manager: DialogManager = kwargs["dialog_manager"]
manager_data = await manager.load_data()
ad_image_url = manager_data["dialog_data"]["campaign"]["ad_image"]
if not ad_image_url:
@@ -109,6 +123,146 @@ async def campaign_by_id(**kwargs: dict[Any]) -> dict[str, Any]:
return {"ad_image": attachment}
async def campaign_on_error(
message: Message,
widget: ManagedTextInput,
dialog_manager: DialogManager,
error: object,
) -> None:
if isinstance(error, ValidationError):
await message.answer(f"Invalid campaign data: {error.json()}")
elif isinstance(error, ValueError):
await message.answer(f"Invalid campaign data {error!s}")
async def campaign_create_on_success(
message: Message,
widget: ManagedTextInput,
dialog_manager: DialogManager,
campaign_data: str,
) -> None:
state = dialog_manager.middleware_data["state"]
state_data = await state.get_data()
fields = message.text.split("\n\n")
campaign = utils.campaign_from_list(fields)
async with AdNovaClient() as client:
try:
await client.create_campaign(
advertiser_id=state_data["advertiser_id"], data=campaign
)
await dialog_manager.switch_to(CampaignsDailogState.campaigns)
except BadRequestError as e:
await message.answer(
f"Invalid data: {e.model_dump(mode='json')['detail']}"
)
async def campaign_edit_on_success(
message: Message,
widget: ManagedTextInput,
dialog_manager: DialogManager,
campaign_data: str,
) -> None:
fields = message.text.split("\n\n")
new_campaign = utils.campaign_from_list(fields)
state = dialog_manager.middleware_data["state"]
state_data = await state.get_data()
manager_data: dict[Any] = await dialog_manager.load_data()
campaign_id = manager_data["dialog_data"]["campaign"]["campaign_id"]
async with AdNovaClient() as client:
try:
new_campaign = await client.update_campaign(
advertiser_id=state_data["advertiser_id"],
campaign_id=campaign_id,
data=new_campaign,
)
await dialog_manager.update(
{
"campaign": new_campaign.model_dump(mode="json"),
}
)
await dialog_manager.switch_to(CampaignsDailogState.campaign)
except BadRequestError as e:
await message.answer(
f"Invalid data: {e.model_dump(mode='json')['detail']}"
)
except ForbiddenError as e:
await message.answer(
"Forbidden changing campaign: "
f"{e.model_dump(mode='json')['detail']}"
)
async def campaign_detail_on_click(
callback: CallbackQuery, button: Button, manager: DialogManager
) -> None:
manager_data: dict[Any] = await manager.load_data()
state: FSMContext = manager_data["middleware_data"]["state"]
state_data = await state.get_data()
advertiser_id = state_data["advertiser_id"]
campaign_id = manager.item_id
if campaign_id == "‎":
return
async with AdNovaClient() as client:
campaign = await client.get_campaign(
advertiser_id=advertiser_id,
campaign_id=campaign_id,
)
campaign_statistics = await client.get_campaign_statistics(
campaign_id=campaign_id
)
if campaign.ad_image:
campaign.ad_image = (
f"{config.MINIO_URL}{urlparse(campaign.ad_image).path}"
)
await manager.update(
{
"campaign": campaign.model_dump(mode="json"),
"campaign_statistics": campaign_statistics.model_dump(mode="json"),
}
)
await callback.answer()
await manager.switch_to(CampaignsDailogState.campaign)
async def campaign_edit_on_click(
callback: CallbackQuery, button: Button, manager: DialogManager
) -> None:
manager_data: dict[Any] = await manager.load_data()
state: FSMContext = manager_data["middleware_data"]["state"]
state_data = await state.get_data()
advertiser_id = state_data["advertiser_id"]
campaign_id = manager_data["dialog_data"]["campaign"]["campaign_id"]
async with AdNovaClient() as client:
campaign = await client.get_campaign(
advertiser_id=advertiser_id,
campaign_id=campaign_id,
)
await manager.update(
{
"campaign": campaign.model_dump(mode="json"),
}
)
await callback.answer()
await manager.switch_to(CampaignsDailogState.campaign_edit)
async def delete_ad_image(
callback: CallbackQuery, button: Button, manager: DialogManager
) -> None:
@@ -150,16 +304,40 @@ async def delete_campaign(
await manager.switch_to(CampaignsDailogState.campaigns)
async def back_to_list(
async def back_to_campaign(
callback: CallbackQuery, button: Button, manager: DialogManager
) -> None:
await manager.switch_to(CampaignsDailogState.campaigns)
manager_data = await manager.load_data()
state: FSMContext = manager_data["middleware_data"]["state"]
state_data = await state.get_data()
campaign = manager_data["dialog_data"]["campaign"]
campaign_id = campaign["campaign_id"]
advertiser_id = state_data["advertiser_id"]
async with AdNovaClient() as client:
campaign = await client.get_campaign(
advertiser_id=advertiser_id,
campaign_id=campaign_id,
)
await manager.update(
{
"campaign": campaign.model_dump(mode="json"),
}
)
await callback.answer()
await manager.switch_to(CampaignsDailogState.campaign)
campaigns_dialog = Dialog(
Window(
Const("Campaigns:"),
Button(Const("➕ Create"), id="create_campaign"),
Start(
Const("➕ Create"),
id="create_campaign",
state=CampaignsDailogState.campaign_create,
),
ScrollingGroup(
ListGroup(
Button(
@@ -178,6 +356,25 @@ campaigns_dialog = Dialog(
state=CampaignsDailogState.campaigns,
getter=campaigns,
),
Window(
Const(
"Enter campaign info in following format "
"(each statement on new line with one enter between each):"
),
*campaign_info,
Start(
Const("⬅️ Back to list"),
id="back",
state=CampaignsDailogState.campaigns,
),
TextInput(
id="campaign",
type_factory=check_campaign,
on_success=campaign_create_on_success,
on_error=campaign_on_error,
),
state=CampaignsDailogState.campaign_create,
),
Window(
DynamicMedia("ad_image", when=campaign_has_ad_image),
Format("• ID: <code>{dialog_data[campaign][campaign_id]}</code>"),
@@ -199,12 +396,36 @@ campaigns_dialog = Dialog(
Format("\t • Age from: {dialog_data[campaign][targeting][age_from]}"),
Format("\t • Age to: {dialog_data[campaign][targeting][age_to]}"),
Format("\t • Location: {dialog_data[campaign][targeting][location]}"),
Button(Const("📝 Edit campaign"), id="edit_campaign"),
Const("\n📊 Statistics:\n"),
Format(
"Impressions: "
"{dialog_data[campaign_statistics][impressions_count]}"
),
Format("Clicks: {dialog_data[campaign_statistics][clicks_count]}"),
Format(
"Conversion: "
"{dialog_data[campaign_statistics][impressions_count]:.2f}%"
),
Format(
"Spent on impressions: "
"{dialog_data[campaign_statistics][impressions_count]:.2f}"
),
Format(
"Spent on clicks: "
"{dialog_data[campaign_statistics][spent_clicks]:.2f}"
),
Format(
"Spent total: {dialog_data[campaign_statistics][spent_total]:.2f}"
),
Button(
Const("📝 Edit campaign"),
id="edit_campaign",
on_click=campaign_edit_on_click,
),
Start(
Const("⬆️ Upload image"),
id="upload_ad_image",
state=CampaignsDailogState.campaign_upload_ad_image,
data=None,
when=campaign_has_not_ad_image,
),
Button(
@@ -218,8 +439,31 @@ campaigns_dialog = Dialog(
id="delete_ad_image",
on_click=delete_campaign,
),
Button(Const("⬅️ Back to list"), id="back", on_click=back_to_list),
Start(
Const("⬅️ Back to list"),
id="back",
state=CampaignsDailogState.campaigns,
),
state=CampaignsDailogState.campaign,
getter=campaign_by_id,
),
Window(
Const(
"Enter new campaign info in following format "
"(each statement on new line with one enter between each):"
),
*campaign_info,
Button(
Const("⬅️ Back to campaign"),
id="back_to_campaign",
on_click=back_to_campaign,
),
TextInput(
id="campaign_update",
type_factory=check_campaign,
on_success=campaign_edit_on_success,
on_error=campaign_on_error,
),
state=CampaignsDailogState.campaign_edit,
),
)
solution/services/telegram_bot/dialogs/start.py
+3 -1
View File
@@ -45,7 +45,9 @@ async def advertiser_id_on_success(
state_data["advertiser_id"] = message.text
await state.set_data(state_data)
await message.answer(f"Successfully authenticated as {message.text}")
await message.answer(
f"Successfully authenticated as {message.text}. Get help: /help."
)
await dialog_manager.mark_closed()
solution/services/telegram_bot/dialogs/utils.py
+20
View File
@@ -0,0 +1,20 @@
from api.schemas import CampaignCreateIn, CampaignTargeting
def campaign_from_list(fields: list[str]) -> CampaignCreateIn:
return CampaignCreateIn(
targeting=CampaignTargeting(
gender=None if fields[8] == "None" else fields[8],
age_from=None if fields[9] == "None" else fields[9],
age_to=None if fields[10] == "None" else fields[10],
location=None if fields[11] == "None" else fields[11],
),
ad_title=fields[0],
ad_text=fields[1],
impressions_limit=fields[2],
clicks_limit=fields[3],
cost_per_impression=fields[4],
cost_per_click=fields[5],
start_date=fields[6],
end_date=fields[7],
)
solution/services/telegram_bot/main.py
+2
View File
@@ -8,6 +8,7 @@ from aiogram_dialog import setup_dialogs
import config
from commands.campaigns import campaigns_router
from commands.help import help_router
from commands.logout import logout_router
from commands.start import start_router
from commands.stats import statistics_router
@@ -43,6 +44,7 @@ async def main() -> None:
dp.message.outer_middleware(AuthMiddleware())
dp.include_routers(
help_router,
start_router,
campaigns_router,
statistics_router,
solution/services/telegram_bot/pyproject.toml
+1
View File
@@ -100,6 +100,7 @@ ignore = [
"FBT001",
"FBT002",
"N813",
"PLR2004",
"RUF001",
"TC002",
]
solution/services/telegram_bot/states/campaigns.py
+1
View File
@@ -5,4 +5,5 @@ class CampaignsDailogState(StatesGroup):
campaigns = State()
campaign = State()
campaign_upload_ad_image = State()
campaign_create = State()
campaign_edit = State()