"""LEGAL DISCLAIMER: This script is provided "as is" as a code sample for demonstration purposes only. Modigie, Inc. makes no representations or warranties of any kind, express or implied, about the completeness, accuracy, reliability, suitability, or availability with respect to this script for any purpose. Any reliance you place on such information is therefore strictly at your own risk. In no event will we be liable for any loss or damage including without limitation, indirect or consequential loss or damage, or any loss or damage whatsoever arising from loss of data or profits arising out of, or in connection with, the use of this script. This script is strictly intended for evaluation batches (<500 records) during onboarding validation sessions and is NOT intended for production use. """ import argparse import csv import importlib.util import json import os import sys import time from typing import Any, Callable, Dict, List, Tuple from google.api_core import exceptions from google.cloud import pubsub try: from . import config except ImportError: try: import config if not hasattr(config, "UtilConfig"): raise ImportError() except ImportError: _cfgPath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.py") _spec = importlib.util.spec_from_file_location("quick_test_util.config", _cfgPath) config = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(config) WARNING_BANNER = """ ******************************************************************************* * WARNING: EVALUATION TOOL ONLY - NOT FIT FOR PRODUCTION * * This utility is strictly designed for one-off quality checks (<500 records) * * during onboarding validation sessions. It lacks production-grade error * * handling, retry backoff, queuing, concurrency, and persistent storage. * ******************************************************************************* """ # --8<-- [start:configuration] # Centralized configuration (can also be overridden via CLI arguments) defaultConfig = config.UtilConfig() # --8<-- [end:configuration] DEFAULT_DELAY_SECONDS = 0.5 DEFAULT_JOB_TYPE = "enrichMobile" DEFAULT_BATCH_ID = "eval" BATCH_ID = DEFAULT_BATCH_ID DEFAULT_CACHE_FILENAME = ".request_cache.json" DEFAULT_CACHE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), DEFAULT_CACHE_FILENAME) def write_request_cache( cacheEntries: Dict[str, Dict[str, Any]], cachePath: str = DEFAULT_CACHE_PATH, ) -> None: """Save or merge request metadata into the local cache file.""" existingCache: Dict[str, Any] = {} if os.path.exists(cachePath): try: with open(cachePath, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict): existingCache = data except Exception: existingCache = {} existingCache.update(cacheEntries) targetDir = os.path.dirname(cachePath) if targetDir: os.makedirs(targetDir, exist_ok=True) with open(cachePath, "w", encoding="utf-8") as f: json.dump(existingCache, f, indent=2, ensure_ascii=False) def clean_row(row: Dict[str, Any]) -> Dict[str, str]: """Strip whitespace from CSV cell values and convert None to empty string.""" return { k: v.strip() if isinstance(v, str) else ("" if v is None else str(v)) for k, v in row.items() if k is not None } def get_default_csv_filename(jobType: str) -> str: """Map job type to its corresponding default CSV filename.""" mapping = { "enrichMobile": "sample_input_enrich_mobile.csv", "verifyEmployment": "sample_input_verify_employment.csv", "verifyMobile": "sample_input_verify_mobile.csv", "enrichLinkedIn": "sample_input_enrich_linkedin.csv", "verifyPhone": "sample_input_verify_phone.csv", } return mapping.get(jobType, f"sample_input_{jobType.lower()}.csv") def build_body_enrich_mobile(row: Dict[str, str]) -> Dict[str, Any]: """Build request payload for enrichMobile job type. Requires: firstName, lastName, company, (email OR linkedInUrl). Optional: recordId, contactId, middleNames, jobTitle, phoneNumber. """ contactId = (row.get("contactId") or row.get("Contact ID") or "").strip() middleNames = (row.get("middleNames") or row.get("middleName") or "").strip() linkedInUrl = (row.get("linkedInUrl") or row.get("linkedinUrl") or "").strip() person: Dict[str, Any] = { "firstName": row.get("firstName", ""), "lastName": row.get("lastName", ""), } if middleNames: person["middleNames"] = middleNames if contactId: person["source"] = { "objectId": contactId, "propertyName": "contactId", } payload: Dict[str, Any] = { "person": person, } if row.get("company") or row.get("jobTitle"): employment: Dict[str, Any] = {} if row.get("jobTitle"): employment["position"] = { "title": row["jobTitle"], "source": {"propertyName": "title"}, } if row.get("company"): employment["company"] = { "title": row["company"], "source": {"propertyName": "company"}, } payload["employment"] = employment if row.get("email"): payload["emailAddresses"] = [{ "emailAddress": {"userFormat": row["email"]}, "rels": ["business"], "source": {"propertyName": "companyEmail"}, }] if linkedInUrl: payload["urls"] = [{ "url": {"userFormat": linkedInUrl}, "source": {"propertyName": "linkedInUrl"}, }] phoneVal = row.get("phoneNumber") if phoneVal: payload["phoneNumbers"] = [{ "phoneNumber": {"userFormat": phoneVal}, "source": {"propertyName": "phone"}, }] return { "payload": payload, "priority": 50, } def build_body_verify_employment(row: Dict[str, str]) -> Dict[str, Any]: """Build request payload for verifyEmployment job type. Requires: company, (firstName + lastName OR linkedInUrl). Optional: recordId, contactId, middleNames, jobTitle, email, phoneNumber. """ contactId = (row.get("contactId") or row.get("Contact ID") or "").strip() middleNames = (row.get("middleNames") or row.get("middleName") or "").strip() linkedInUrl = (row.get("linkedInUrl") or row.get("linkedinUrl") or "").strip() payload: Dict[str, Any] = {} if row.get("company") or row.get("jobTitle"): employment: Dict[str, Any] = {} if row.get("jobTitle"): employment["position"] = { "title": row["jobTitle"], "source": {"propertyName": "title"}, } if row.get("company"): employment["company"] = { "title": row["company"], "source": {"propertyName": "company"}, } payload["employment"] = employment if row.get("firstName") or row.get("lastName") or middleNames: person = { "firstName": row.get("firstName", ""), "lastName": row.get("lastName", ""), } if middleNames: person["middleNames"] = middleNames if contactId: person["source"] = { "objectId": contactId, "propertyName": "contactId", } payload["person"] = person if linkedInUrl: payload["urls"] = [{ "url": {"userFormat": linkedInUrl}, "source": {"propertyName": "linkedInUrl"}, }] if row.get("email"): payload["emailAddresses"] = [{ "emailAddress": {"userFormat": row["email"]}, "rels": ["business"], "source": {"propertyName": "companyEmail"}, }] phoneVal = row.get("phoneNumber") if phoneVal: payload["phoneNumbers"] = [{ "phoneNumber": {"userFormat": phoneVal}, "source": {"propertyName": "phone"}, }] return { "payload": payload, "priority": 50, } def build_body_verify_mobile(row: Dict[str, str]) -> Dict[str, Any]: """Build request payload for verifyMobile job type. Requires: firstName, lastName, phoneNumber (E.164). Optional: recordId, contactId, middleNames, company, jobTitle, email, linkedInUrl. """ contactId = (row.get("contactId") or row.get("Contact ID") or "").strip() middleNames = (row.get("middleNames") or row.get("middleName") or "").strip() linkedInUrl = (row.get("linkedInUrl") or row.get("linkedinUrl") or "").strip() person: Dict[str, Any] = { "firstName": row.get("firstName", ""), "lastName": row.get("lastName", ""), } if middleNames: person["middleNames"] = middleNames if contactId: person["source"] = { "objectId": contactId, "propertyName": "contactId", } payload: Dict[str, Any] = { "person": person, } phoneVal = row.get("phoneNumber") if phoneVal: payload["phoneNumbers"] = [{ "phoneNumber": {"userFormat": phoneVal}, "source": {"propertyName": "mobile"}, }] if row.get("company") or row.get("jobTitle"): employment: Dict[str, Any] = {} if row.get("jobTitle"): employment["position"] = { "title": row["jobTitle"], "source": {"propertyName": "title"}, } if row.get("company"): employment["company"] = { "title": row["company"], "source": {"propertyName": "company"}, } payload["employment"] = employment if row.get("email"): payload["emailAddresses"] = [{ "emailAddress": {"userFormat": row["email"]}, "rels": ["business"], "source": {"propertyName": "companyEmail"}, }] if linkedInUrl: payload["urls"] = [{ "url": {"userFormat": linkedInUrl}, "source": {"propertyName": "linkedInUrl"}, }] return { "payload": payload, "priority": 50, } def build_body_enrich_linkedin(row: Dict[str, str]) -> Dict[str, Any]: """Build request payload for enrichLinkedIn job type. Requires: firstName, lastName, company, email. Optional: recordId, contactId, middleNames, jobTitle, phoneNumber. """ contactId = (row.get("contactId") or row.get("Contact ID") or "").strip() middleNames = (row.get("middleNames") or row.get("middleName") or "").strip() person: Dict[str, Any] = { "firstName": row.get("firstName", ""), "lastName": row.get("lastName", ""), } if middleNames: person["middleNames"] = middleNames if contactId: person["source"] = { "objectId": contactId, "propertyName": "contactId", } payload: Dict[str, Any] = { "person": person, } if row.get("company") or row.get("jobTitle"): employment: Dict[str, Any] = {} if row.get("jobTitle"): employment["position"] = { "title": row["jobTitle"], "source": {"propertyName": "title"}, } if row.get("company"): employment["company"] = { "title": row["company"], "source": {"propertyName": "company"}, } payload["employment"] = employment if row.get("email"): payload["emailAddresses"] = [{ "emailAddress": {"userFormat": row["email"]}, "rels": ["business"], "source": {"propertyName": "companyEmail"}, }] phoneVal = row.get("phoneNumber") if phoneVal: payload["phoneNumbers"] = [{ "phoneNumber": {"userFormat": phoneVal}, "source": {"propertyName": "phone"}, }] return { "payload": payload, "priority": 50, } def build_body_verify_phone(row: Dict[str, str]) -> Dict[str, Any]: """Build request payload for verifyPhone job type. Requires: phoneNumber (E.164). Optional: recordId, contactId, firstName, middleNames, lastName, company, linkedInUrl. """ contactId = (row.get("contactId") or row.get("Contact ID") or "").strip() middleNames = (row.get("middleNames") or row.get("middleName") or "").strip() linkedInUrl = (row.get("linkedInUrl") or row.get("linkedinUrl") or "").strip() payload: Dict[str, Any] = {} phoneVal = row.get("phoneNumber") if phoneVal: payload["phoneNumbers"] = [{ "phoneNumber": {"userFormat": phoneVal}, "source": {"propertyName": "phone"}, }] if row.get("firstName") or row.get("lastName") or middleNames: person = { "firstName": row.get("firstName", ""), "lastName": row.get("lastName", ""), } if middleNames: person["middleNames"] = middleNames if contactId: person["source"] = { "objectId": contactId, "propertyName": "contactId", } payload["person"] = person if row.get("company"): payload["employment"] = { "company": { "title": row["company"], "source": {"propertyName": "company"}, } } if linkedInUrl: payload["urls"] = [{ "url": {"userFormat": linkedInUrl}, "source": {"propertyName": "linkedInUrl"}, }] return { "payload": payload, "priority": 50, } BUILDERS: Dict[str, Callable[[Dict[str, str]], Dict[str, Any]]] = { "enrichMobile": build_body_enrich_mobile, "verifyEmployment": build_body_verify_employment, "verifyMobile": build_body_verify_mobile, "enrichLinkedIn": build_body_enrich_linkedin, "verifyPhone": build_body_verify_phone, } def data_scanner( jobType: str, row: Dict[str, str], indexStr: str = "1", batchId: str = BATCH_ID, ) -> Tuple[str, Dict[str, Any]]: r"""Build a serialized job request payload from a CSV row and compute its requestId. The computed requestId forms the modigieJobRequestId attribute (regex: ^[a-zA-Z0-9\-._]{1,256}$): - batchId: forms the beginning and must start with an alphanumeric character [a-zA-Z0-9]. - recordId / contactId / indexStr: form the remainder after the prefix, allowing [a-zA-Z0-9\-._]. Args: jobType: Target Modigie job type. row: Dictionary representing a cleaned CSV row. indexStr: Formatted index string of the row with leading zeros. batchId: Batch identifier prefix. Returns: Tuple[str, Dict[str, Any]]: (requestId, requestBody) """ if jobType not in BUILDERS: raise ValueError(f"No payload builder registered for job type '{jobType}'") recordId = (row.get("recordId") or row.get("Record ID") or "").strip() contactId = (row.get("contactId") or row.get("Contact ID") or "").strip() if recordId: requestId = f"{batchId}_{recordId}" if batchId else recordId elif row.get("requestId"): requestId = row["requestId"] else: requestId = f"{batchId}_{indexStr}" if batchId else indexStr rowCopy = dict(row) rowCopy["requestId"] = requestId if contactId: rowCopy["contactId"] = contactId builder = BUILDERS[jobType] requestBody = builder(rowCopy) tags = [] if batchId: tags.append({"name": "modigieBatch", "value": batchId}) if recordId: tags.append({"name": "recordId", "value": recordId}) else: tags.append({"name": "recordId", "value": str(indexStr)}) if contactId: tags.append({"name": "contactId", "value": contactId}) requestBody.setdefault("payload", {}) if "tags" in requestBody["payload"]: requestBody["payload"]["tags"].extend(tags) else: requestBody["payload"]["tags"] = tags return requestId, requestBody def file_scanner( csvPath: str, jobType: str, batchId: str = BATCH_ID, saveCache: bool = False, cachePath: str = DEFAULT_CACHE_PATH, ) -> List[Tuple[str, Dict[str, Any]]]: """Read a CSV file and parse each row into (requestId, requestBody) tuples. Args: csvPath: Path to input CSV file. jobType: Target Modigie job type. batchId: Batch identifier prefix. saveCache: Whether to write row metadata to the local request cache. cachePath: Destination path for the cache file. Returns: List of tuples: (requestId, requestBody) """ if not os.path.exists(csvPath): raise FileNotFoundError(f"CSV file not found at: {csvPath}") records: List[Tuple[str, Dict[str, Any]]] = [] cacheEntries: Dict[str, Dict[str, Any]] = {} with open(csvPath, "r", encoding="utf-8") as f: firstLine = f.readline() delimiter = "\t" if "\t" in firstLine else "," f.seek(0) reader = csv.DictReader(f, delimiter=delimiter) rows = list(reader) totalRows = len(rows) indexWidth = len(str(totalRows)) if totalRows > 0 else 1 for rowIdx, rawRow in enumerate(rows, start=1): row = clean_row(rawRow) indexStr = f"{rowIdx:0{indexWidth}d}" requestId, requestBody = data_scanner( jobType=jobType, row=row, indexStr=indexStr, batchId=batchId, ) records.append((requestId, requestBody)) recId = (row.get("recordId") or row.get("Record ID") or "").strip() or str(indexStr) cntId = (row.get("contactId") or row.get("Contact ID") or "").strip() rowCache = dict(row) rowCache["recordId"] = recId rowCache["contactId"] = cntId cacheEntries[requestId] = rowCache if saveCache and cacheEntries: write_request_cache(cacheEntries, cachePath=cachePath) return records def publish_from_csv( csvPath: str, jobType: str, jobConfig: config.UtilConfig, delaySeconds: float = DEFAULT_DELAY_SECONDS, isDryRun: bool = False, batchId: str = BATCH_ID, cachePath: str = DEFAULT_CACHE_PATH, ) -> None: """Read a CSV file and publish one Pub/Sub message per row for the configured job type.""" print(WARNING_BANNER) if jobType not in BUILDERS: raise ValueError(f"No payload builder registered for job type '{jobType}'") topicPath = jobConfig.get_topic_path(jobType) print(f"Active Job Type: {jobType}") print(f"Target Topic: {topicPath}") print(f"Input CSV File: {csvPath}") print(f"Dry-run Mode: {isDryRun}\n") if not os.path.exists(csvPath): print(f"Error: CSV file not found: {csvPath}") sys.exit(1) records = file_scanner(csvPath=csvPath, jobType=jobType, batchId=batchId, saveCache=True, cachePath=cachePath) totalRead = len(records) publishedCount = 0 previousFuture = None publisherClient = None if not isDryRun: creds = jobConfig.get_credentials() publisherClient = pubsub.PublisherClient(credentials=creds) for idx, (requestId, requestBody) in enumerate(records, start=1): if not isDryRun and previousFuture is not None and delaySeconds > 0: time.sleep(delaySeconds) requestData = json.dumps(requestBody, ensure_ascii=False).encode("utf-8") messageAttributes = {"modigieJobRequestId": requestId} if isDryRun: print(f"--- [DRY-RUN] Row {idx}: requestId={requestId} ---") print(f"Attributes: {messageAttributes}") print(f"Body:\n{json.dumps(requestBody, indent=2, ensure_ascii=False)}\n") publishedCount += 1 else: try: future = publisherClient.publish(topicPath, requestData, **messageAttributes) messageId = future.result() print(f"Published {requestId} -> messageId={messageId}") previousFuture = future publishedCount += 1 except exceptions.PermissionDenied as exc: print(f"\n[Permission Denied: 403] Error publishing {requestId}: {exc.message or exc}") print("\nTip: Publishing requires 'roles/pubsub.publisher' on the topic in tenant project.") print("When running in a Cloud Workstation, ensure the workstation's default service account") print("has been granted the required Pub/Sub IAM roles for this tenant repository.") print("For local environments using a service account key, set:") print("export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json\n") sys.exit(1) modeDesc = "Simulated publish for" if isDryRun else "Published" print(f"\nDone. {modeDesc} {publishedCount} job(s) from {totalRead} rows (skipped: 0).") def main() -> None: """CLI entrypoint for publisher.""" parser = argparse.ArgumentParser(description="Publish Modigie job requests to Cloud Pub/Sub from a CSV file.") parser.add_argument( "csv_pos", nargs="?", default=None, help="Optional positional path to input CSV file.", ) parser.add_argument( "--csv", dest="csv_flag", default=None, help="Path to input CSV file. Defaults to sample_input_{jobType}.csv", ) parser.add_argument( "--jobType", default=None, choices=list(config.SUPPORTED_JOB_TYPES), help=f"Target Modigie job type. Defaults to {DEFAULT_JOB_TYPE}", ) parser.add_argument( "--tenantProjectId", "--projectId", dest="tenantProjectId", default=None, help=f"GCP Project ID. Defaults to '{defaultConfig.tenantProjectId}'", ) parser.add_argument( "--repoShortId", "--repoHash", dest="repoShortId", default=None, help=f"Repository short ID (15 hex chars). Defaults to '{defaultConfig.repoShortId}'", ) parser.add_argument( "--delay", type=float, default=DEFAULT_DELAY_SECONDS, help=f"Interval in seconds between published messages (default: {DEFAULT_DELAY_SECONDS}).", ) parser.add_argument( "--batchId", dest="batchId", default=None, help=( "Batch ID prefix for computed requestId. Must start with an alphanumeric character " f"[a-zA-Z0-9] and contain only [a-zA-Z0-9-._] (defaults to '{DEFAULT_BATCH_ID}')." ), ) parser.add_argument( "--dryRun", "--preview", dest="dryRun", action="store_true", help="Preview generated JSON payloads without publishing to Pub/Sub.", ) parser.add_argument( "--credentials", "--keyFile", dest="credentialsPath", default=None, help="Optional path to Google Cloud service account JSON key file (overrides MODIGIE_CREDENTIALS_JSON).", ) args = parser.parse_args() activeJobType = args.jobType or DEFAULT_JOB_TYPE activeBatchId = args.batchId or DEFAULT_BATCH_ID jobConfig = config.UtilConfig( tenantProjectId=args.tenantProjectId, repoShortId=args.repoShortId, credentialsPath=args.credentialsPath, ) csvPath = args.csv_flag or args.csv_pos if not csvPath: defaultCsvName = get_default_csv_filename(activeJobType) scriptDir = os.path.dirname(os.path.abspath(__file__)) csvPath = os.path.join(scriptDir, defaultCsvName) publish_from_csv( csvPath=csvPath, jobType=activeJobType, jobConfig=jobConfig, delaySeconds=args.delay, isDryRun=args.dryRun, batchId=activeBatchId, ) if __name__ == "__main__": main()