Product Reviews Management

class ProductReviewsServiceClient:
    def __init__(
        self,
        *,
        credentials: Optional[ga_credentials.Credentials] = None,
        transport: Optional[Union[str, ProductReviewsServiceTransport, Callable]] = None,
        client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
        client_info: gapic_v1.client_info.ClientInfo = None
    ):
        """
        Initialize the product reviews service client.
        
        Args:
            credentials: Google authentication credentials
            transport: Transport to use for communication ('grpc', 'grpc_asyncio', 'rest', or transport instance)
            client_options: Client configuration options
            client_info: Client information for user agent
        """

class ProductReviewsServiceAsyncClient:
    def __init__(
        self,
        *,
        credentials: Optional[ga_credentials.Credentials] = None,
        transport: Optional[Union[str, ProductReviewsServiceTransport, Callable]] = None,
        client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
        client_info: gapic_v1.client_info.ClientInfo = None
    ):
        """
        Initialize the async product reviews service client.
        
        Args:
            credentials: Google authentication credentials
            transport: Transport to use for communication ('grpc_asyncio' or transport instance)
            client_options: Client configuration options
            client_info: Client information for user agent
        """

def get_product_review(
    self,
    request: Optional[Union[GetProductReviewRequest, dict]] = None,
    *,
    name: Optional[str] = None,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ProductReview:
    """
    Gets a product review.
    
    Args:
        request: The request object or dict containing the request fields
        name: Required. Resource name format: accounts/{account}/productReviews/{productReview}
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata for the request
        
    Returns:
        ProductReview: The requested product review
        
    Raises:
        google.api_core.exceptions.NotFound: If the review is not found
        google.api_core.exceptions.PermissionDenied: If access is denied
    """

async def get_product_review(
    self,
    request: Optional[Union[GetProductReviewRequest, dict]] = None,
    *,
    name: Optional[str] = None,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ProductReview:
    """Async version of get_product_review."""

def list_product_reviews(
    self,
    request: Optional[Union[ListProductReviewsRequest, dict]] = None,
    *,
    parent: Optional[str] = None,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ListProductReviewsPager:
    """
    Lists product reviews for an account.
    
    Args:
        request: The request object or dict containing the request fields
        parent: Required. Parent resource format: accounts/{account}
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata for the request
        
    Returns:
        ListProductReviewsPager: Pager for iterating through reviews
        
    Raises:
        google.api_core.exceptions.InvalidArgument: If parent format is invalid
        google.api_core.exceptions.PermissionDenied: If access is denied
    """

async def list_product_reviews(
    self,
    request: Optional[Union[ListProductReviewsRequest, dict]] = None,
    *,
    parent: Optional[str] = None,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ListProductReviewsAsyncPager:
    """Async version of list_product_reviews."""

def insert_product_review(
    self,
    request: Optional[Union[InsertProductReviewRequest, dict]] = None,
    *,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ProductReview:
    """
    Inserts a product review.
    
    Args:
        request: The request object containing parent, product_review, and data_source
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata for the request
        
    Returns:
        ProductReview: The inserted product review
        
    Raises:
        google.api_core.exceptions.InvalidArgument: If request data is invalid
        google.api_core.exceptions.PermissionDenied: If access is denied
    """

async def insert_product_review(
    self,
    request: Optional[Union[InsertProductReviewRequest, dict]] = None,
    *,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ProductReview:
    """Async version of insert_product_review."""

def delete_product_review(
    self,
    request: Optional[Union[DeleteProductReviewRequest, dict]] = None,
    *,
    name: Optional[str] = None,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> None:
    """
    Deletes a product review.
    
    Args:
        request: The request object or dict containing the request fields
        name: Required. Resource name format: accounts/{account}/productReviews/{productReview}
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata for the request
        
    Raises:
        google.api_core.exceptions.NotFound: If the review is not found
        google.api_core.exceptions.PermissionDenied: If access is denied
    """

async def delete_product_review(
    self,
    request: Optional[Union[DeleteProductReviewRequest, dict]] = None,
    *,
    name: Optional[str] = None,
    retry = gapic_v1.method.DEFAULT,
    timeout = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> None:
    """Async version of delete_product_review."""

@classmethod
def product_review_path(cls, account: str, name: str) -> str:
    """
    Returns a fully-qualified product review string.
    
    Args:
        account: Account ID
        name: Product review name/ID
        
    Returns:
        str: Resource path in format accounts/{account}/productReviews/{productReview}
    """

@classmethod
def parse_product_review_path(cls, path: str) -> Dict[str, str]:
    """
    Parses a product review path.
    
    Args:
        path: Product review resource path
        
    Returns:
        Dict[str, str]: Dictionary with 'account' and 'product_review' keys
        
    Raises:
        ValueError: If path format is invalid
    """

@classmethod
def from_service_account_file(
    cls, 
    filename: str, 
    *args, 
    **kwargs
) -> ProductReviewsServiceClient:
    """
    Creates a client from a service account file.
    
    Args:
        filename: Path to service account JSON file
        *args: Additional arguments to pass to constructor
        **kwargs: Additional keyword arguments to pass to constructor
        
    Returns:
        ProductReviewsServiceClient: Configured client instance
    """

@classmethod
def from_service_account_info(
    cls,
    info: dict,
    *args,
    **kwargs
) -> ProductReviewsServiceClient:
    """
    Creates a client from service account info.
    
    Args:
        info: Service account info dictionary
        *args: Additional arguments to pass to constructor
        **kwargs: Additional keyword arguments to pass to constructor
        
    Returns:
        ProductReviewsServiceClient: Configured client instance
    """

from google.shopping.merchant_reviews import (
    ProductReviewsServiceClient,
    ProductReview,
    ProductReviewAttributes,
    InsertProductReviewRequest
)
from google.auth import default
from google.protobuf.timestamp_pb2 import Timestamp
import time

# Initialize client
credentials, project = default()
client = ProductReviewsServiceClient(credentials=credentials)

# Create a new product review
review_attributes = ProductReviewAttributes(
    reviewer_username="happy_customer",
    reviewer_id="reviewer456",
    content="This product exceeded my expectations! Great quality and fast shipping.",
    title="Excellent product!",
    rating=5.0,
    min_rating=1,
    max_rating=5,
    product_names=["Wireless Bluetooth Headphones"],
    brands=["AudioTech"],
    gtins=["1234567890123"],
    is_verified_purchase=True,
    collection_method=ProductReviewAttributes.CollectionMethod.POST_FULFILLMENT
)

# Set review time
review_time = Timestamp()
review_time.FromSeconds(int(time.time()))
review_attributes.review_time = review_time

product_review = ProductReview(
    product_review_id="product-review-789",
    product_review_attributes=review_attributes
)

# Insert the review
request = InsertProductReviewRequest(
    parent="accounts/your-account-id",
    product_review=product_review,
    data_source="accounts/your-account-id/dataSources/your-datasource-id"
)

result = client.insert_product_review(request=request)
print(f"Inserted product review: {result.name}")

# List product reviews
reviews = client.list_product_reviews(parent="accounts/your-account-id")
for review in reviews:
    print(f"Product Review ID: {review.product_review_id}")
    print(f"Rating: {review.product_review_attributes.rating}")
    print(f"Product: {review.product_review_attributes.product_names[0] if review.product_review_attributes.product_names else 'N/A'}")

# Get specific product review
review_name = f"accounts/your-account-id/productReviews/product-review-789"
specific_review = client.get_product_review(name=review_name)
print(f"Retrieved product review rating: {specific_review.product_review_attributes.rating}")

# Delete product review
client.delete_product_review(name=review_name)
print("Product review deleted")

# Create comprehensive product review with multiple attributes
review_attributes = ProductReviewAttributes(
    aggregator_name="ReviewPlatform",
    publisher_name="TechReviews Inc",
    reviewer_username="tech_expert_2024",
    reviewer_id="expert789",
    title="Comprehensive Analysis of Wireless Headphones",
    content="After using these headphones for 3 months, I can confidently say they deliver exceptional audio quality...",
    pros=["Excellent sound quality", "Long battery life", "Comfortable fit"],
    cons=["Could use better noise cancellation", "Price is a bit high"],
    rating=4.2,
    min_rating=1,
    max_rating=5,
    product_names=["Premium Wireless Bluetooth Headphones", "Model XYZ-2024"],
    product_links=["https://example.com/product/xyz-2024"],
    brands=["AudioTech", "PremiumSound"],
    gtins=["1234567890123"],
    mpns=["XYZ-2024-BLK"],
    skus=["AUDIO-XYZ-BLK"],
    asins=["B08EXAMPLE"],
    is_verified_purchase=True,
    is_spam=False,
    is_incentivized_review=False,
    collection_method=ProductReviewAttributes.CollectionMethod.POST_FULFILLMENT,
    transaction_id="txn-abc123456",
    review_language="en",
    review_country="US"
)

# Set review time
review_time = Timestamp()
review_time.FromSeconds(int(time.time() - 86400))  # 1 day ago
review_attributes.review_time = review_time

product_review = ProductReview(
    product_review_id="comprehensive-review-001",
    product_review_attributes=review_attributes
)

# Insert comprehensive review
request = InsertProductReviewRequest(
    parent="accounts/your-account-id",
    product_review=product_review,
    data_source="accounts/your-account-id/dataSources/comprehensive-reviews"
)

result = client.insert_product_review(request=request)
print(f"Inserted comprehensive product review: {result.name}")

import asyncio
from google.shopping.merchant_reviews import ProductReviewsServiceAsyncClient

async def analyze_product_reviews():
    client = ProductReviewsServiceAsyncClient()
    
    # Get all product reviews asynchronously
    reviews = await client.list_product_reviews(
        parent="accounts/your-account-id"
    )
    
    # Calculate average rating
    total_rating = 0
    count = 0
    low_rated_reviews = []
    
    async for review in reviews:
        rating = review.product_review_attributes.rating
        if rating > 0:  # Only count reviews with ratings
            total_rating += rating
            count += 1
            
            if rating < 3.0:
                low_rated_reviews.append(review)
    
    if count > 0:
        average_rating = total_rating / count
        print(f"Average product rating: {average_rating:.2f}")
        print(f"Total reviews: {count}")
        print(f"Low-rated reviews: {len(low_rated_reviews)}")
        
        # Process low-rated reviews
        for review in low_rated_reviews:
            print(f"Low rating alert: {review.product_review_id} - {review.product_review_attributes.rating}")

# Run async analysis
asyncio.run(analyze_product_reviews())

# Request types
class GetProductReviewRequest:
    name: str  # Required: accounts/{account}/productReviews/{productReview}

class DeleteProductReviewRequest:
    name: str  # Required: accounts/{account}/productReviews/{productReview}

class ListProductReviewsRequest:
    parent: str  # Required: accounts/{account}
    page_size: int  # Optional: maximum results per page
    page_token: str  # Optional: pagination token

class InsertProductReviewRequest:
    parent: str  # Required: accounts/{account}
    product_review: ProductReview  # Required: review to insert
    data_source: str  # Required: accounts/{account}/dataSources/{datasource}

# Response types
class ListProductReviewsResponse:
    product_reviews: MutableSequence[ProductReview]
    next_page_token: str
    @property
    def raw_page(self) -> ListProductReviewsResponse: ...