diff --git a/src/adloop/ads/write.py b/src/adloop/ads/write.py index a1847a1..f771376 100644 --- a/src/adloop/ads/write.py +++ b/src/adloop/ads/write.py @@ -515,6 +515,140 @@ def add_to_negative_keyword_list( return plan.to_preview() +def _normalize_shared_set_attachment_args( + shared_set_id: str, + campaign_ids: list[str] | None, +) -> tuple[list[str], list[str]]: + """Validate inputs for attach/detach. Returns (errors, deduped_campaign_ids).""" + errors: list[str] = [] + if not shared_set_id: + errors.append("shared_set_id is required") + elif not str(shared_set_id).isdigit(): + errors.append( + "shared_set_id must be a numeric ID (from get_negative_keyword_lists)" + ) + + campaign_ids = campaign_ids or [] + if not campaign_ids: + errors.append("At least one campaign_id is required") + + seen: set[str] = set() + deduped: list[str] = [] + for cid in campaign_ids: + cid_str = str(cid).strip() + if not cid_str: + continue + if not cid_str.isdigit(): + errors.append(f"campaign_id '{cid_str}' must be numeric") + continue + if cid_str in seen: + continue + seen.add(cid_str) + deduped.append(cid_str) + + return errors, deduped + + +def attach_shared_set_to_campaigns( + config: AdLoopConfig, + *, + customer_id: str = "", + shared_set_id: str = "", + campaign_ids: list[str] | None = None, +) -> dict: + """Draft attaching an existing shared set to one or more campaigns — returns PREVIEW. + + Creates ``CampaignSharedSet`` linkages so the campaigns inherit the shared + set's criteria (e.g. negative keywords). Most commonly used to attach a + shared negative keyword list to newly-built campaigns. Use + ``get_negative_keyword_lists`` to find the ``shared_set_id`` and + ``get_negative_keyword_list_campaigns`` to see existing attachments. + + shared_set_id: numeric ID of the shared set to attach. + campaign_ids: list of numeric campaign IDs to attach the set to. Duplicates + in the input list are collapsed. + + Call ``confirm_and_apply`` with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("attach_shared_set_to_campaigns", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + errors, deduped = _normalize_shared_set_attachment_args( + shared_set_id, campaign_ids + ) + if errors: + return {"error": "Validation failed", "details": errors} + + plan = ChangePlan( + operation="attach_shared_set_to_campaigns", + entity_type="campaign_shared_set", + entity_id=str(shared_set_id), + customer_id=customer_id, + changes={ + "shared_set_id": str(shared_set_id), + "campaign_ids": deduped, + }, + ) + store_plan(plan) + return plan.to_preview() + + +def detach_shared_set_from_campaigns( + config: AdLoopConfig, + *, + customer_id: str = "", + shared_set_id: str = "", + campaign_ids: list[str] | None = None, +) -> dict: + """Draft detaching a shared set from one or more campaigns — returns PREVIEW. + + Removes ``CampaignSharedSet`` linkages so the campaigns no longer inherit + the shared set's criteria. The shared set itself is unchanged; only the + per-campaign attachment record is removed. Use + ``get_negative_keyword_list_campaigns`` to inspect existing attachments + before detaching. + + shared_set_id: numeric ID of the shared set. + campaign_ids: list of numeric campaign IDs to detach the set from. + Detaching a set that isn't currently attached to a campaign is a no-op + at the API level (the request will fail for that specific operation + but does not affect the others; surfaced in the apply response). + + Call ``confirm_and_apply`` with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("detach_shared_set_from_campaigns", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + errors, deduped = _normalize_shared_set_attachment_args( + shared_set_id, campaign_ids + ) + if errors: + return {"error": "Validation failed", "details": errors} + + plan = ChangePlan( + operation="detach_shared_set_from_campaigns", + entity_type="campaign_shared_set", + entity_id=str(shared_set_id), + customer_id=customer_id, + changes={ + "shared_set_id": str(shared_set_id), + "campaign_ids": deduped, + }, + ) + store_plan(plan) + return plan.to_preview() + + def update_ad_group( config: AdLoopConfig, *, @@ -1922,6 +2056,8 @@ def _execute_plan(config: AdLoopConfig, plan: object) -> dict: "add_negative_keywords": _apply_add_negative_keywords, "create_negative_keyword_list": _apply_create_negative_keyword_list, "add_to_negative_keyword_list": _apply_add_to_negative_keyword_list, + "attach_shared_set_to_campaigns": _apply_attach_shared_set_to_campaigns, + "detach_shared_set_from_campaigns": _apply_detach_shared_set_from_campaigns, "pause_entity": _apply_status_change, "enable_entity": _apply_status_change, "remove_entity": _apply_remove, @@ -2847,3 +2983,172 @@ def _apply_add_to_negative_keyword_list( "resource_names": [r.resource_name for r in response.results], "keyword_count": len(response.results), } + + +def _parse_partial_failure_per_op( + client: object, partial_failure_error: object +) -> dict: + """Parse a Google Ads partial_failure_error proto into {op_index: message}. + + Returns an empty dict when there are no failures, when ``details`` is + missing, or when proto deserialization fails for any reason — callers + can still detect failed operations via empty ``result.resource_name`` + entries and surface ``partial_failure_error.message`` as a fallback. + """ + if partial_failure_error is None: + return {} + if not getattr(partial_failure_error, "code", 0): + return {} + + details = getattr(partial_failure_error, "details", None) or [] + out: dict = {} + try: + failure_msg = client.get_type("GoogleAdsFailure") + failure_pb_cls = type(failure_msg).pb(failure_msg).__class__ + for detail in details: + value = getattr(detail, "value", None) + if value is None: + continue + failure = failure_pb_cls.FromString(value) + for err in failure.errors: + fpe = getattr(err.location, "field_path_elements", None) + if fpe: + out[fpe[0].index] = err.message + except Exception: + pass + return out + + +def _apply_attach_shared_set_to_campaigns( + client: object, cid: str, changes: dict +) -> dict: + """Create CampaignSharedSet linkages for one shared set across campaigns. + + Uses ``partial_failure=True`` so per-operation errors (e.g. attempting + to attach a list to a campaign that already has it) don't fail the + whole batch. Successful linkages are returned in ``resource_names``; + failed linkages are surfaced under ``failed_campaigns`` with the + originating campaign_id and per-op error message when parseable. + """ + css_service = client.get_service("CampaignSharedSetService") + campaign_service = client.get_service("CampaignService") + shared_set_service = client.get_service("SharedSetService") + + shared_set_resource = shared_set_service.shared_set_path( + cid, changes["shared_set_id"] + ) + + campaign_ids = list(changes["campaign_ids"]) + operations = [] + for campaign_id in campaign_ids: + op = client.get_type("CampaignSharedSetOperation") + css = op.create + css.campaign = campaign_service.campaign_path(cid, campaign_id) + css.shared_set = shared_set_resource + operations.append(op) + + response = css_service.mutate_campaign_shared_sets( + customer_id=cid, + operations=operations, + partial_failure=True, + ) + + pf_error = getattr(response, "partial_failure_error", None) + per_op_errors = _parse_partial_failure_per_op(client, pf_error) + + succeeded: list = [] + failed: list = [] + for idx, result in enumerate(response.results): + if result.resource_name: + succeeded.append(result.resource_name) + else: + failed.append( + { + "campaign_id": str(campaign_ids[idx]), + "operation_index": idx, + "error": per_op_errors.get( + idx, "Unknown error (see partial_failure_message)" + ), + } + ) + + out = { + "shared_set_resource": shared_set_resource, + "resource_names": succeeded, + "campaign_count": len(succeeded), + } + if failed: + out["partial_failure"] = True + out["failed_campaigns"] = failed + if pf_error is not None: + msg = getattr(pf_error, "message", "") + if msg: + out["partial_failure_message"] = msg + return out + + +def _apply_detach_shared_set_from_campaigns( + client: object, cid: str, changes: dict +) -> dict: + """Remove CampaignSharedSet linkages for one shared set across campaigns. + + The CampaignSharedSet resource name has a composite ID of the form + ``{campaign_id}~{shared_set_id}`` so we can construct the resource path + directly without needing to look up the linkage first. + + Uses ``partial_failure=True`` so per-operation errors (e.g. detaching + a non-existent linkage) don't fail the whole batch. Successful removals + are returned in ``removed_resource_names``; failed operations are + surfaced under ``failed_campaigns`` with the originating campaign_id + and per-op error message when parseable. + """ + css_service = client.get_service("CampaignSharedSetService") + + shared_set_id = changes["shared_set_id"] + campaign_ids = list(changes["campaign_ids"]) + operations = [] + for campaign_id in campaign_ids: + op = client.get_type("CampaignSharedSetOperation") + op.remove = ( + f"customers/{cid}/campaignSharedSets/{campaign_id}~{shared_set_id}" + ) + operations.append(op) + + response = css_service.mutate_campaign_shared_sets( + customer_id=cid, + operations=operations, + partial_failure=True, + ) + + pf_error = getattr(response, "partial_failure_error", None) + per_op_errors = _parse_partial_failure_per_op(client, pf_error) + + succeeded: list = [] + failed: list = [] + for idx, result in enumerate(response.results): + if result.resource_name: + succeeded.append(result.resource_name) + else: + failed.append( + { + "campaign_id": str(campaign_ids[idx]), + "operation_index": idx, + "error": per_op_errors.get( + idx, "Unknown error (see partial_failure_message)" + ), + } + ) + + out = { + "shared_set_id": shared_set_id, + "removed_resource_names": succeeded, + "campaign_count": len(succeeded), + } + if failed: + out["partial_failure"] = True + out["failed_campaigns"] = failed + if pf_error is not None: + msg = getattr(pf_error, "message", "") + if msg: + out["partial_failure_message"] = msg + return out diff --git a/src/adloop/server.py b/src/adloop/server.py index fb9035a..71c07cc 100644 --- a/src/adloop/server.py +++ b/src/adloop/server.py @@ -1050,6 +1050,68 @@ def add_to_negative_keyword_list( ) +@mcp.tool(annotations=_WRITE) +@_safe +def attach_shared_set_to_campaigns( + shared_set_id: str, + campaign_ids: list[str], + customer_id: str = "", +) -> dict: + """Attach an existing shared set to one or more campaigns — returns a PREVIEW. + + Creates CampaignSharedSet linkages so the campaigns inherit the shared + set's criteria (e.g. negative keywords). Most commonly used to attach a + shared negative keyword list to newly-built campaigns. + + Use ``get_negative_keyword_lists`` to find the shared_set_id, and + ``get_negative_keyword_list_campaigns`` to inspect existing attachments. + + shared_set_id: numeric ID from get_negative_keyword_lists. + campaign_ids: list of numeric campaign IDs to attach the set to. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import attach_shared_set_to_campaigns as _impl + + return _impl( + _config, + customer_id=customer_id or _config.ads.customer_id, + shared_set_id=shared_set_id, + campaign_ids=campaign_ids, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def detach_shared_set_from_campaigns( + shared_set_id: str, + campaign_ids: list[str], + customer_id: str = "", +) -> dict: + """Detach a shared set from one or more campaigns — returns a PREVIEW. + + Removes CampaignSharedSet linkages so the campaigns no longer inherit the + shared set's criteria. The shared set itself is unchanged; only the + per-campaign attachment is removed. + + Use ``get_negative_keyword_list_campaigns`` to inspect existing attachments + before detaching. + + shared_set_id: numeric ID from get_negative_keyword_lists. + campaign_ids: list of numeric campaign IDs to detach the set from. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import detach_shared_set_from_campaigns as _impl + + return _impl( + _config, + customer_id=customer_id or _config.ads.customer_id, + shared_set_id=shared_set_id, + campaign_ids=campaign_ids, + ) + + @mcp.tool(annotations=_WRITE) @_safe def update_ad_group( diff --git a/tests/test_ads_write.py b/tests/test_ads_write.py index ce333fe..e76090c 100644 --- a/tests/test_ads_write.py +++ b/tests/test_ads_write.py @@ -1415,3 +1415,270 @@ def test_description_length_check_still_enforced_on_dict_entries(self): "https://example.com/", ) assert any("exceeds 90 chars" in e for e in errs) + + +# --------------------------------------------------------------------------- +# attach_shared_set_to_campaigns / detach_shared_set_from_campaigns +# --------------------------------------------------------------------------- + + +class TestAttachSharedSetValidation: + def test_rejects_missing_shared_set_id(self, config): + result = write.attach_shared_set_to_campaigns( + config, customer_id="123-456-7890", campaign_ids=["1001"] + ) + assert result["error"] == "Validation failed" + assert "shared_set_id is required" in result["details"] + + def test_rejects_non_numeric_shared_set_id(self, config): + result = write.attach_shared_set_to_campaigns( + config, + customer_id="123-456-7890", + shared_set_id="abc", + campaign_ids=["1001"], + ) + assert result["error"] == "Validation failed" + assert any("shared_set_id" in d and "numeric" in d for d in result["details"]) + + def test_rejects_empty_campaign_ids(self, config): + result = write.attach_shared_set_to_campaigns( + config, + customer_id="123-456-7890", + shared_set_id="9001", + campaign_ids=[], + ) + assert result["error"] == "Validation failed" + assert "At least one campaign_id is required" in result["details"] + + def test_rejects_non_numeric_campaign_id(self, config): + result = write.attach_shared_set_to_campaigns( + config, + customer_id="123-456-7890", + shared_set_id="9001", + campaign_ids=["1001", "abc"], + ) + assert result["error"] == "Validation failed" + assert any("'abc'" in d and "numeric" in d for d in result["details"]) + + def test_dedups_campaign_ids(self, config): + result = write.attach_shared_set_to_campaigns( + config, + customer_id="123-456-7890", + shared_set_id="9001", + campaign_ids=["1001", "1002", "1001", " 1003 "], + ) + assert result["operation"] == "attach_shared_set_to_campaigns" + assert result["changes"]["campaign_ids"] == ["1001", "1002", "1003"] + + def test_returns_preview_with_plan_id(self, config): + result = write.attach_shared_set_to_campaigns( + config, + customer_id="123-456-7890", + shared_set_id="9001", + campaign_ids=["1001"], + ) + assert "plan_id" in result + assert result["entity_type"] == "campaign_shared_set" + assert result["entity_id"] == "9001" + assert result["changes"]["shared_set_id"] == "9001" + + +class TestDetachSharedSetValidation: + def test_rejects_missing_shared_set_id(self, config): + result = write.detach_shared_set_from_campaigns( + config, customer_id="123-456-7890", campaign_ids=["1001"] + ) + assert result["error"] == "Validation failed" + assert "shared_set_id is required" in result["details"] + + def test_returns_preview_with_plan_id(self, config): + result = write.detach_shared_set_from_campaigns( + config, + customer_id="123-456-7890", + shared_set_id="9001", + campaign_ids=["1001", "1002"], + ) + assert "plan_id" in result + assert result["operation"] == "detach_shared_set_from_campaigns" + assert result["entity_type"] == "campaign_shared_set" + assert result["changes"]["campaign_ids"] == ["1001", "1002"] + + +def test_apply_attach_shared_set_to_campaigns_creates_one_op_per_campaign(): + captured: dict = {} + + def _mutate(customer_id, operations, partial_failure=False): + captured["customer_id"] = customer_id + captured["operations"] = list(operations) + captured["partial_failure"] = partial_failure + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/campaignSharedSets/{i}~9001" + ) + for i, _ in enumerate(operations, start=1001) + ], + partial_failure_error=SimpleNamespace(code=0, message="", details=[]), + ) + + css_service = SimpleNamespace(mutate_campaign_shared_sets=_mutate) + campaign_service = _FakePathService("campaigns") + shared_set_service = SimpleNamespace( + shared_set_path=lambda cid, ssid: f"customers/{cid}/sharedSets/{ssid}" + ) + + client = _FakeClient( + { + "CampaignSharedSetService": css_service, + "CampaignService": campaign_service, + "SharedSetService": shared_set_service, + } + ) + + result = write._apply_attach_shared_set_to_campaigns( + client, + "1234567890", + {"shared_set_id": "9001", "campaign_ids": ["1001", "1002", "1003"]}, + ) + + assert captured["customer_id"] == "1234567890" + assert captured["partial_failure"] is True + assert len(captured["operations"]) == 3 + op0 = captured["operations"][0] + assert op0.create.campaign == "customers/1234567890/campaigns/1001" + assert op0.create.shared_set == "customers/1234567890/sharedSets/9001" + assert result["campaign_count"] == 3 + assert len(result["resource_names"]) == 3 + assert "partial_failure" not in result + assert "failed_campaigns" not in result + + +def test_apply_attach_shared_set_to_campaigns_surfaces_partial_failure(): + """When a subset of attach ops fail, succeeded ones return resource_names + and failed ones surface in failed_campaigns with their campaign_id.""" + + def _mutate(customer_id, operations, partial_failure=False): + # Op #1 (index 1, campaign_id "1002") fails — empty resource_name. + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/campaignSharedSets/1001~9001" + ), + SimpleNamespace(resource_name=""), + SimpleNamespace( + resource_name=f"customers/{customer_id}/campaignSharedSets/1003~9001" + ), + ], + partial_failure_error=SimpleNamespace( + code=3, + message="Some operations failed", + details=[], + ), + ) + + css_service = SimpleNamespace(mutate_campaign_shared_sets=_mutate) + campaign_service = _FakePathService("campaigns") + shared_set_service = SimpleNamespace( + shared_set_path=lambda cid, ssid: f"customers/{cid}/sharedSets/{ssid}" + ) + + client = _FakeClient( + { + "CampaignSharedSetService": css_service, + "CampaignService": campaign_service, + "SharedSetService": shared_set_service, + } + ) + + result = write._apply_attach_shared_set_to_campaigns( + client, + "1234567890", + {"shared_set_id": "9001", "campaign_ids": ["1001", "1002", "1003"]}, + ) + + assert result["campaign_count"] == 2 + assert result["resource_names"] == [ + "customers/1234567890/campaignSharedSets/1001~9001", + "customers/1234567890/campaignSharedSets/1003~9001", + ] + assert result["partial_failure"] is True + assert result["partial_failure_message"] == "Some operations failed" + assert len(result["failed_campaigns"]) == 1 + failed = result["failed_campaigns"][0] + assert failed["campaign_id"] == "1002" + assert failed["operation_index"] == 1 + assert "error" in failed + + +def test_apply_detach_shared_set_from_campaigns_uses_composite_resource_name(): + captured: dict = {} + + def _mutate(customer_id, operations, partial_failure=False): + captured["customer_id"] = customer_id + captured["operations"] = list(operations) + captured["partial_failure"] = partial_failure + return SimpleNamespace( + results=[ + SimpleNamespace(resource_name=op.remove) for op in operations + ], + partial_failure_error=SimpleNamespace(code=0, message="", details=[]), + ) + + css_service = SimpleNamespace(mutate_campaign_shared_sets=_mutate) + client = _FakeClient({"CampaignSharedSetService": css_service}) + + result = write._apply_detach_shared_set_from_campaigns( + client, + "1234567890", + {"shared_set_id": "9001", "campaign_ids": ["1001", "1002"]}, + ) + + assert captured["customer_id"] == "1234567890" + assert captured["partial_failure"] is True + ops = captured["operations"] + assert len(ops) == 2 + assert ops[0].remove == "customers/1234567890/campaignSharedSets/1001~9001" + assert ops[1].remove == "customers/1234567890/campaignSharedSets/1002~9001" + assert result["shared_set_id"] == "9001" + assert result["campaign_count"] == 2 + assert "partial_failure" not in result + assert "failed_campaigns" not in result + + +def test_apply_detach_shared_set_from_campaigns_surfaces_partial_failure(): + """When a detach op targets a non-existent linkage, that op fails but + the others succeed and the result surfaces both.""" + + def _mutate(customer_id, operations, partial_failure=False): + # Op #0 (campaign_id "1001") fails — empty resource_name. + return SimpleNamespace( + results=[ + SimpleNamespace(resource_name=""), + SimpleNamespace(resource_name=operations[1].remove), + ], + partial_failure_error=SimpleNamespace( + code=3, + message="Some operations failed", + details=[], + ), + ) + + css_service = SimpleNamespace(mutate_campaign_shared_sets=_mutate) + client = _FakeClient({"CampaignSharedSetService": css_service}) + + result = write._apply_detach_shared_set_from_campaigns( + client, + "1234567890", + {"shared_set_id": "9001", "campaign_ids": ["1001", "1002"]}, + ) + + assert result["campaign_count"] == 1 + assert result["removed_resource_names"] == [ + "customers/1234567890/campaignSharedSets/1002~9001", + ] + assert result["partial_failure"] is True + assert result["partial_failure_message"] == "Some operations failed" + assert len(result["failed_campaigns"]) == 1 + failed = result["failed_campaigns"][0] + assert failed["campaign_id"] == "1001" + assert failed["operation_index"] == 0