Implement assets (#881)

* Update packages

* Implement sitelinks assets

* Implement callout assets

* Implement client function google_ads_create_assets

captn/google_ads/client.py

@@ -4,6 +4,7 @@
 
 from pydantic import BaseModel
 from requests import get as requests_get
+from requests import post as requests_post
 
 BASE_URL = environ.get("CAPTN_BACKEND_URL", "http://localhost:9000")
 ALREADY_AUTHENTICATED = "User is already authenticated"
@@ -291,20 +292,18 @@ def check_for_client_approval(
     return NOT_APPROVED
 
 
-def google_ads_create_update(
+def _check_for_approval_and_get_login_url(
     user_id: int,
     conv_id: int,
-    ad: BaseModel,
+    model: BaseModel,
     recommended_modifications_and_answer_list: List[
         Tuple[Dict[str, Any], Optional[str]]
     ],
-    login_customer_id: Optional[str] = None,
-    endpoint: str = "/update-ad-group-ad",
-    already_checked_clients_approval: bool = False,
-) -> Union[Dict[str, Any], str]:
+    already_checked_clients_approval: bool,
+) -> Optional[Dict[str, str]]:
     if not already_checked_clients_approval:
         error_msg = check_for_client_approval(
-            modification_function_parameters=ad.model_dump(),
+            modification_function_parameters=model.model_dump(),
             recommended_modifications_and_answer_list=recommended_modifications_and_answer_list,
         )
         if error_msg:
@@ -314,6 +313,30 @@ def google_ads_create_update(
     if not login_url_response.get("login_url") == ALREADY_AUTHENTICATED:
         return login_url_response
 
+    return None
+
+
+def google_ads_create_update(
+    user_id: int,
+    conv_id: int,
+    ad: BaseModel,
+    recommended_modifications_and_answer_list: List[
+        Tuple[Dict[str, Any], Optional[str]]
+    ],
+    login_customer_id: Optional[str] = None,
+    endpoint: str = "/update-ad-group-ad",
+    already_checked_clients_approval: bool = False,
+) -> Union[Dict[str, Any], str]:
+    login_url_response = _check_for_approval_and_get_login_url(
+        user_id=user_id,
+        conv_id=conv_id,
+        model=ad,
+        recommended_modifications_and_answer_list=recommended_modifications_and_answer_list,
+        already_checked_clients_approval=already_checked_clients_approval,
+    )
+    if login_url_response:
+        return login_url_response
+
     params: Dict[str, Any] = ad.model_dump()
     params["user_id"] = user_id
     params["login_customer_id"] = login_customer_id
@@ -324,3 +347,40 @@ def google_ads_create_update(
 
     response_dict: Union[Dict[str, Any], str] = response.json()
     return response_dict
+
+
+def google_ads_create_assets(
+    user_id: int,
+    conv_id: int,
+    model: BaseModel,
+    recommended_modifications_and_answer_list: List[
+        Tuple[Dict[str, Any], Optional[str]]
+    ],
+    endpoint: str,
+    already_checked_clients_approval: bool = False,
+) -> Union[Dict[str, Any], str]:
+    login_url_response = _check_for_approval_and_get_login_url(
+        user_id=user_id,
+        conv_id=conv_id,
+        model=model,
+        recommended_modifications_and_answer_list=recommended_modifications_and_answer_list,
+        already_checked_clients_approval=already_checked_clients_approval,
+    )
+    if login_url_response:
+        return login_url_response
+
+    params = {
+        "user_id": user_id,
+    }
+
+    body = model.model_dump()
+
+    response = requests_post(
+        f"{BASE_URL}{endpoint}", json=body, params=params, timeout=60
+    )
+
+    if not response.ok:
+        raise ValueError(response.content)
+
+    response_dict: Union[Dict[str, Any], str] = response.json()
+    return response_dict

google_ads/application.py

@@ -25,8 +25,10 @@
     AdGroupAd,
     AdGroupCriterion,
     Campaign,
+    CampaignCallouts,
     CampaignCriterion,
     CampaignLanguageCriterion,
+    CampaignSitelinks,
     Criterion,
     GeoTargetCriterion,
     RemoveResource,
@@ -1603,3 +1605,128 @@ async def remove_google_ads_resource(
         ) from e
 
     return f"Removed {response.results[0].resource_name}."
+
+
+def _link_assets_to_campaign(
+    client: Any,
+    customer_id: str,
+    campaign_id: str,
+    resource_names: List[str],
+    field_type: Any,
+) -> None:
+    campaign_service = client.get_service("CampaignService")
+    operations = []
+    for resource_name in resource_names:
+        operation = client.get_type("CampaignAssetOperation")
+        campaign_asset = operation.create
+        campaign_asset.asset = resource_name
+        campaign_asset.campaign = campaign_service.campaign_path(
+            customer_id, campaign_id
+        )
+        campaign_asset.field_type = field_type
+        # campaign_asset.field_type = client.enums.AssetFieldTypeEnum.CALLOUT
+        operations.append(operation)
+
+    campaign_asset_service = client.get_service("CampaignAssetService")
+    campaign_asset_service.mutate_campaign_assets(
+        customer_id=customer_id, operations=operations
+    )
+
+
+def _create_sitelink_operations(client: Any, model: CampaignSitelinks) -> List[Any]:
+    operations = []
+    for site_link in model.site_links:
+        operation = client.get_type("AssetOperation")
+        asset = operation.create
+        asset.final_urls.extend(site_link.final_urls)
+        if site_link.description1:
+            asset.sitelink_asset.description1 = site_link.description1
+        if site_link.description2:
+            asset.sitelink_asset.description2 = site_link.description2
+        asset.sitelink_asset.link_text = site_link.link_text
+        operations.append(operation)
+    return operations
+
+
+def _create_callout_operations(client: Any, model: CampaignCallouts) -> List[Any]:
+    operations = []
+    for callout in model.callouts:
+        operation = client.get_type("AssetOperation")
+        asset = operation.create
+        asset.callout_asset.callout_text = callout
+        operations.append(operation)
+    return operations
+
+
+def _create_assets(
+    client: Any, model: Union[CampaignSitelinks, CampaignCallouts]
+) -> List[str]:
+    if isinstance(model, CampaignSitelinks):
+        operations = _create_sitelink_operations(client, model)
+    else:
+        operations = _create_callout_operations(client, model)
+
+    asset_service = client.get_service("AssetService")
+    response = asset_service.mutate_assets(
+        customer_id=model.customer_id,
+        operations=operations,
+    )
+
+    resource_names = [result.resource_name for result in response.results]
+
+    return resource_names
+
+
+def _create_assets_helper(
+    client: Any,
+    model: Union[CampaignSitelinks, CampaignCallouts],
+    field_type: Any,
+) -> str:
+    try:
+        resource_names = _create_assets(client, model)
+        _link_assets_to_campaign(
+            client=client,
+            customer_id=model.customer_id,
+            campaign_id=model.campaign_id,
+            resource_names=resource_names,
+            field_type=field_type,
+        )
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)
+        ) from e
+    return (
+        f"Linked following assets to campaign '{model.campaign_id}': {resource_names}"
+    )
+
+
+@router.post("/create-sitelinks-for-campaign")
+async def create_sitelinks_for_campaign(
+    user_id: int,
+    model: CampaignSitelinks,
+) -> str:
+    client = await _get_client(
+        user_id=user_id, login_customer_id=model.login_customer_id
+    )
+
+    result = _create_assets_helper(
+        client=client, model=model, field_type=client.enums.AssetFieldTypeEnum.SITELINK
+    )
+
+    return result
+
+
+@router.post("/create-callouts-for-campaign")
+async def create_callouts_for_campaign(
+    user_id: int,
+    model: CampaignCallouts,
+) -> str:
+    client = await _get_client(
+        user_id=user_id, login_customer_id=model.login_customer_id
+    )
+
+    result = _create_assets_helper(
+        client=client, model=model, field_type=client.enums.AssetFieldTypeEnum.CALLOUT
+    )
+
+    return result

google_ads/model.py

@@ -1,7 +1,7 @@
-from typing import List, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional
 
 from fastapi import Query
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 class AdBase(BaseModel):
@@ -157,3 +157,47 @@ class GeoTargetCriterion(BaseModel):
     negative: Optional[bool] = None
     target_type: Optional[Literal["Country", "County", "City", "Region"]] = None
     add_all_suggestions: Optional[bool] = None
+
+
+class SiteLink(BaseModel):
+    final_urls: List[str]
+    link_text: str = Field(max_length=25)
+    description1: Optional[str] = Field(default=None, max_length=35)
+    description2: Optional[str] = Field(default=None, max_length=35)
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_descriptions(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+        description1 = values.get("description1")
+        description2 = values.get("description2")
+
+        if (description1 is None) != (description2 is None):
+            raise ValueError(
+                "Either both description1 and description2 should be provided, or neither."
+            )
+
+        return values
+
+
+class CampaignSitelinks(BaseModel):
+    customer_id: str
+    login_customer_id: Optional[str] = None
+    campaign_id: str
+    site_links: List[SiteLink]
+
+
+class CampaignCallouts(BaseModel):
+    customer_id: str
+    login_customer_id: Optional[str] = None
+    campaign_id: str
+    callouts: List[str]
+
+    @field_validator("callouts")
+    def callouts_validator(cls, callouts: List[str]) -> List[str]:
+        error_message = check_max_string_length_for_each_item(
+            field_name="callouts", field=callouts, max_string_length=25
+        )
+
+        if error_message:
+            raise ValueError(error_message)
+        return callouts

pyproject.toml

@@ -97,7 +97,7 @@ agents = [
     "httpx==0.27.0",
     "uvicorn==0.30.5",
     "python-dotenv==1.0.1",
-    "pyautogen[websurfer,websockets]==0.2.32",
+    "pyautogen[websurfer,websockets,anthropic,together]==0.2.34",
     "pandas>=2.1",
     "fastcore==1.6.7",
     "asyncer==0.0.7",

tests/ci/google_ads/test_model.py

@@ -1,7 +1,14 @@
+from typing import Optional
+
 import pytest
 from pydantic import ValidationError
 
-from google_ads.model import AdGroupAd, _remove_keyword_insertion_chars
+from google_ads.model import (
+    AdGroupAd,
+    CampaignCallouts,
+    SiteLink,
+    _remove_keyword_insertion_chars,
+)
 
 
 class TestAdGroupAd:
@@ -71,3 +78,67 @@ def test_maximum_headline_string_length(self, headlines, expected):
                 headlines=headlines,
                 descriptions=["Description 1", "Description 2"],
             )
+
+
+class TestSiteLink:
+    def test_link_text_longer_than_25_characters(self) -> None:
+        with pytest.raises(ValueError) as exc_info:
+            SiteLink(link_text="A" * 26, final_urls=["https://www.example.com"])
+
+        assert "String should have at most 25 characters" in str(exc_info.value)
+
+    @pytest.mark.parametrize(
+        ("description1", "description2", "expected"),
+        [
+            ("Description 1", "Description 2", None),
+            ("Description 1", None, ValueError),
+            (None, "Description 2", ValueError),
+            (None, None, None),
+        ],
+    )
+    def test_descritptions(
+        self, description1: str, description2: str, expected: Optional[Exception]
+    ) -> None:
+        if expected is not None:
+            with pytest.raises(ValueError) as exc_info:
+                SiteLink(
+                    link_text="Link Text",
+                    final_urls=["https://www.example.com"],
+                    description1=description1,
+                    description2=description2,
+                )
+            assert (
+                "Either both description1 and description2 should be provided, or neither"
+                in str(exc_info.value)
+            )
+        else:
+            SiteLink(
+                link_text="Link Text",
+                final_urls=["https://www.example.com"],
+                description1=description1,
+                description2=description2,
+            )
+
+
+class TestCampaignCallouts:
+    @pytest.mark.parametrize(
+        "callouts, expected",
+        [
+            (["Callout 1", "Callout 2", "Callout 3"], None),
+            (["Callout 1", "Callout 2", "C" * 26], ValueError),
+        ],
+    )
+    def test_maximum_callout_string_length(self, callouts, expected):
+        if expected is not None:
+            with pytest.raises(ValueError):
+                CampaignCallouts(
+                    customer_id="2222",
+                    campaign_id="3333",
+                    callouts=callouts,
+                )
+        else:
+            CampaignCallouts(
+                customer_id="2222",
+                campaign_id="3333",
+                callouts=callouts,
+            )