Add page feed items endpoints

captn/captn_agents/backend/tools/_gbb_google_sheets_team_tools.py

@@ -30,7 +30,7 @@
 from ....google_ads.client import (
     execute_query,
     google_ads_create_update,
-    google_ads_post,
+    google_ads_post_or_get,
 )
 from ....google_ads.client import (
     list_accessible_customers_with_account_types as list_accessible_customers_with_account_types_client,
@@ -296,7 +296,7 @@ def _add_negative_campaign_keywords_lists(
             campaign_id=campaign_id,
             shared_set_name=row["keyword"],
         )
-        google_ads_post(
+        google_ads_post_or_get(
             user_id=context.user_id,
             conv_id=context.conv_id,
             recommended_modifications_and_answer_list=context.recommended_modifications_and_answer_list,
@@ -505,7 +505,7 @@ def _update_callouts(
         callouts=callouts,
     )
 
-    response = google_ads_post(
+    response = google_ads_post_or_get(
         user_id=context.user_id,
         conv_id=context.conv_id,
         recommended_modifications_and_answer_list=context.recommended_modifications_and_answer_list,
@@ -636,7 +636,7 @@ def _add_existing_sitelinks(
         sitelink_ids=sitelink_ids,
     )
 
-    google_ads_post(
+    google_ads_post_or_get(
         user_id=context.user_id,
         conv_id=context.conv_id,
         recommended_modifications_and_answer_list=context.recommended_modifications_and_answer_list,
@@ -687,7 +687,7 @@ def _add_new_sitelinks(
         site_links=site_links,
     )
 
-    google_ads_post(
+    google_ads_post_or_get(
         user_id=context.user_id,
         conv_id=context.conv_id,
         recommended_modifications_and_answer_list=context.recommended_modifications_and_answer_list,

captn/google_ads/client.py

@@ -1,6 +1,6 @@
 import json
 from os import environ
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Literal, Optional, Tuple, Union
 
 from pydantic import BaseModel
 from requests import get as requests_get
@@ -351,7 +351,7 @@ def google_ads_create_update(
     return response_dict
 
 
-def google_ads_post(
+def google_ads_post_or_get(
     user_id: int,
     conv_id: int,
     model: BaseModel,
@@ -360,6 +360,7 @@ def google_ads_post(
     ],
     endpoint: str,
     already_checked_clients_approval: bool = False,
+    requests_method: Literal["get", "post"] = "post",
 ) -> Union[Dict[str, Any], str]:
     login_url_response = _check_for_approval_and_get_login_url(
         user_id=user_id,
@@ -377,7 +378,8 @@ def google_ads_post(
 
     body = model.model_dump()
 
-    response = requests_post(
+    requests_methods = {"get": requests_get, "post": requests_post}
+    response = requests_methods[requests_method](  # type: ignore[operator]
         f"{BASE_URL}{endpoint}", json=body, params=params, timeout=60
     )
 

google_ads/application.py

@@ -21,6 +21,7 @@
 from .model import (
     AdBase,
     AdCopy,
+    AddPageFeedItems,
     AdGroup,
     AdGroupAd,
     AdGroupCriterion,
@@ -33,6 +34,7 @@
     ExistingCampaignSitelinks,
     GeoTargetCriterion,
     NewCampaignSitelinks,
+    PageFeedItems,
     RemoveResource,
 )
 
@@ -931,6 +933,12 @@ def _create_ad_group_ad_set_attr(
         "setattr_create_func": _keywords_setattr,
         "setattr_update_func": _set_fields,
     },
+    "asset_set_asset": {
+        "service": "AssetSetAssetService",
+        "operation": "AssetSetAssetOperation",
+        "mutate": "mutate_asset_set_assets",
+        "service_path_update_delete": "asset_set_asset_path",
+    },
 }
 
 
@@ -1955,3 +1963,122 @@ async def add_shared_set_to_campaign(
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)
         ) from e
+
+
+def _add_assets_to_asset_set(
+    client: Any,
+    customer_id: str,
+    asset_resource_names: List[str],
+    asset_set_resource_name: str,
+) -> str:
+    """Adds assets to an asset set by creating an asset set asset link.
+
+    Args:
+        client: an initialized GoogleAdsClient instance.
+        customer_id: a client customer ID.
+        asset_resource_names: a list of asset resource names.
+        asset_set_resource_name: a resource name for an asset set.
+    """
+    operations = []
+    for resource_name in asset_resource_names:
+        # Creates an asset set asset operation and adds it to the list of
+        # operations.
+        operation = client.get_type("AssetSetAssetOperation")
+        asset_set_asset = operation.create
+        asset_set_asset.asset = resource_name
+        asset_set_asset.asset_set = asset_set_resource_name
+        operations.append(operation)
+
+    # Issues a mutate request to add the asset set assets and prints its
+    # information.
+    asset_set_asset_service = client.get_service("AssetSetAssetService")
+    response = asset_set_asset_service.mutate_asset_set_assets(
+        customer_id=customer_id, operations=operations
+    )
+
+    return_text = ""
+    for result in response.results:
+        return_text += f"Created an asset set asset link with resource name '{result.resource_name}'\n"
+    return return_text
+
+
+@router.post("/add-items-to-page-feed")
+async def add_items_to_page_feed(
+    user_id: int,
+    model: AddPageFeedItems,
+) -> str:
+    client = await _get_client(
+        user_id=user_id, login_customer_id=model.login_customer_id
+    )
+    operations = []
+
+    # Creates one asset per URL.
+    for url, label in model.urls_and_labels.items():
+        # Creates an asset operation and adds it to the list of operations.
+        operation = client.get_type("AssetOperation")
+        asset = operation.create
+        page_feed_asset = asset.page_feed_asset
+        page_feed_asset.page_url = url
+        # Recommended: adds labels to the asset. These labels can be used later
+        # in ad group targeting to restrict the set of pages that can serve.
+        if label:
+            page_feed_asset.labels.append(label)
+        operations.append(operation)
+
+    # Issues a mutate request to add the assets and prints its information.
+    asset_service = client.get_service("AssetService")
+    response = asset_service.mutate_assets(
+        customer_id=model.customer_id, operations=operations
+    )
+
+    resource_names = []
+    return_text = ""
+    for result in response.results:
+        resource_name = result.resource_name
+        return_text += f"Created an asset with resource name: '{resource_name}'\n"
+        resource_names.append(resource_name)
+
+    return _add_assets_to_asset_set(
+        client=client,
+        customer_id=model.customer_id,
+        asset_resource_names=resource_names,
+        asset_set_resource_name=model.asset_set_resource_name,
+    )
+
+
+@router.get("/list-page-feed-items")
+async def list_page_feed_items(
+    user_id: int,
+    model: PageFeedItems,
+) -> Dict[str, Any]:
+    # client = await _get_client(
+    #     user_id=user_id, login_customer_id=model.login_customer_id
+    # )
+    query = f"""
+SELECT
+#   asset_set_asset.asset,
+#   asset_set_asset.asset_set,
+  asset.id,
+  asset.name,
+  asset.type,
+  asset.page_feed_asset.page_url
+FROM
+  asset_set_asset
+WHERE
+  asset.type = 'PAGE_FEED'
+  AND asset_set_asset.asset_set = '{model.asset_set_resource_name}'
+  AND asset_set_asset.status != 'REMOVED'
+"""  # nosec: [B608]
+
+    page_feed_assets_response = await search(
+        user_id=user_id,
+        customer_ids=[model.customer_id],
+        query=query,
+        login_customer_id=model.login_customer_id,
+    )
+
+    # page_urls = []
+    # for asset in page_feed_assets_response[model.customer_id]:
+    #     page_urls.append(asset["asset"]["pageFeedAsset"]["pageUrl"])
+
+    return page_feed_assets_response

google_ads/model.py

@@ -146,6 +146,7 @@ class RemoveResource(BaseModel):
         "ad",
         "ad_group_criterion",
         "campaign_criterion",
+        "asset_set_asset",
     ]
 
 
@@ -215,3 +216,13 @@ class CampaignSharedSet(BaseModel):
     customer_id: str
     campaign_id: str
     shared_set_name: str
+
+
+class PageFeedItems(BaseModel):
+    login_customer_id: Optional[str] = None
+    customer_id: str
+    asset_set_resource_name: str
+
+
+class AddPageFeedItems(PageFeedItems):
+    urls_and_labels: Dict[str, Optional[str]]