"""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 re import threading from concurrent.futures import TimeoutError as FuturesTimeoutError from typing import Any, Dict, List, Optional, Set, 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) JOB_ID_PATTERN = re.compile(r"^[a-zA-Z0-9\-._]{1,256}$") 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_OUTPUT_DIR = "responses" DEFAULT_JOB_TYPE = "enrichMobile" DEFAULT_CACHE_FILENAME = ".request_cache.json" DEFAULT_CACHE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), DEFAULT_CACHE_FILENAME) def load_request_cache(cachePath: str = DEFAULT_CACHE_PATH) -> Dict[str, Dict[str, Any]]: """Load cached request row data written by publisher if available.""" if not os.path.isfile(cachePath): return {} try: with open(cachePath, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else {} except Exception: return {} # Only terminal job responses and request discards are considered final/last messages to be written to CSV. TERMINAL_JOB_STATUSES = frozenset({"completed", "rejected", "canceled", "failed"}) _csvLock = threading.Lock() _seenLock = threading.Lock() CSV_HEADERS_BY_JOB_TYPE: Dict[str, List[str]] = { "verifyMobile": [ "requestId", "status", "recordId", "contactId", "reasonType", "reasonTitle", "reasonBody", "firstName", "middleNames", "lastName", "personIsValid", "eventsVerifyMobileCode", "verifiedMobilePhoneType", "verifiedMobileE164Format", "verifiedMobileIntlFormat", "verifiedMobileNatlFormat", "verifiedMobileIsValid", "verifiedMobileIsPossible", "verifiedMobileCountryAlpha2", "verifiedMobileCountryTitle", "verifiedMobileItuCountryCode", "verifiedMobileRatingScore", "verifiedMobileRatingNumberActivityScore", "verifiedMobileRatingNumberActivityCode", "verifiedMobileRatingNumberNameMatchScore", "verifiedMobileRatingNumberNameMatchCode", "validatedEmailNormalFormat", "validatedEmailIsSyntaxValid", "validatedEmailIsDeliverable", "validatedLinkedInNormalFormat", "validatedLinkedInIsSyntaxValid", "validatedLinkedInIsRemote", "validatedLinkedInRels", ], "verifyEmployment": [ "requestId", "status", "recordId", "contactId", "reasonType", "reasonTitle", "reasonBody", "firstName", "middleNames", "lastName", "personIsValid", "eventsVerifyEmploymentCode", "verifiedEmploymentCurrentEmployer", "verifiedEmploymentCurrentPosition", "verifiedEmploymentLeftTargetCompany", "locationsFormattedAddress", "validatedEmailNormalFormat", "validatedEmailIsSyntaxValid", "validatedEmailIsDeliverable", "validatedLinkedInNormalFormat", "validatedLinkedInIsSyntaxValid", "validatedLinkedInIsRemote", "validatedLinkedInRels", "appendedLinkedInNormalFormat", "replacedLinkedInNormalFormat", ], "enrichMobile": [ "requestId", "status", "recordId", "contactId", "reasonType", "reasonTitle", "reasonBody", "firstName", "middleNames", "lastName", "personIsValid", "eventsVerifyEmploymentCode", "verifiedEmploymentCurrentEmployer", "verifiedEmploymentCurrentPosition", "verifiedEmploymentLeftTargetCompany", "locationsFormattedAddress", "eventsAppendMobileCode", "appendedMobilePhoneType", "appendedMobileE164Format", "appendedMobileIntlFormat", "appendedMobileNatlFormat", "appendedMobileCountryAlpha2", "appendedMobileCountryTitle", "appendedMobileItuCountryCode", "appendedMobileRatingScore", "appendedMobileRatingNumberActivityScore", "appendedMobileRatingNumberActivityCode", "appendedMobileRatingNumberNameMatchScore", "appendedMobileRatingNumberNameMatchCode", "validatedEmailNormalFormat", "validatedEmailIsSyntaxValid", "validatedEmailIsDeliverable", "validatedLinkedInNormalFormat", "validatedLinkedInIsSyntaxValid", "validatedLinkedInIsRemote", "validatedLinkedInRels", "appendedLinkedInNormalFormat", "replacedLinkedInNormalFormat", ], "verifyPhone": [ "requestId", "status", "recordId", "contactId", "reasonType", "reasonTitle", "reasonBody", "firstName", "middleNames", "lastName", "personIsValid", "eventsVerifyPhoneCode", "verifiedPhonePhoneType", "verifiedPhoneE164Format", "verifiedPhoneIntlFormat", "verifiedPhoneNatlFormat", "verifiedPhoneIsValid", "verifiedPhoneIsPossible", "verifiedPhoneCountryAlpha2", "verifiedPhoneCountryTitle", "verifiedPhoneItuCountryCode", "verifiedPhoneRatingScore", "verifiedPhoneRatingNumberActivityScore", "verifiedPhoneRatingNumberActivityCode", "verifiedPhoneRatingNumberNameMatchScore", "verifiedPhoneRatingNumberNameMatchCode", "validatedEmailNormalFormat", "validatedEmailIsSyntaxValid", "validatedEmailIsDeliverable", "validatedLinkedInNormalFormat", "validatedLinkedInIsSyntaxValid", "validatedLinkedInIsRemote", "validatedLinkedInRels", ], "enrichLinkedIn": [ "requestId", "status", "recordId", "contactId", "reasonType", "reasonTitle", "reasonBody", "firstName", "middleNames", "lastName", "personIsValid", "eventsAppendLinkedInCode", "appendedLinkedInNormalFormat", "validatedEmailNormalFormat", "validatedEmailIsSyntaxValid", "validatedEmailIsDeliverable", ], } def get_csv_headers(jobType: str) -> List[str]: """Validate the job type and return its ordered list of column headers.""" if jobType not in CSV_HEADERS_BY_JOB_TYPE: raise ValueError(f"Unsupported job type: '{jobType}'. Supported: {', '.join(CSV_HEADERS_BY_JOB_TYPE.keys())}") return list(CSV_HEADERS_BY_JOB_TYPE[jobType]) def safe_get(data: Any, *keys: Any, default: Any = "") -> Any: """Safely traverse nested dictionaries and lists.""" current = data for key in keys: if isinstance(current, dict): current = current.get(key) elif isinstance(current, list) and isinstance(key, int): if 0 <= key < len(current): current = current[key] else: return default else: return default if current is None: return default return current def format_bool(val: Any) -> str: """Format boolean value as 'true' or 'false', returning empty string if None or not a bool.""" if val is None or val == "": return "" if isinstance(val, bool): return "true" if val else "false" if isinstance(val, str): if val.lower() in ("true", "1"): return "true" if val.lower() in ("false", "0"): return "false" return "" def extract_reason(body: Dict[str, Any]) -> str: """Pull a short human-readable reason out of a response body, if present.""" reason = body.get("reason") if isinstance(reason, dict): return str(reason.get("body") or reason.get("message") or json.dumps(reason)) if isinstance(reason, str): return reason return "" def _flatten_phone_details( row: Dict[str, Any], events: Dict[str, Any], primaryPhone: Dict[str, Any], ) -> None: """Populate phone and rating details across appendMobile, verifyMobile, and verifyPhone.""" phoneObj = primaryPhone.get("phoneNumber") if isinstance(primaryPhone.get("phoneNumber"), dict) else {} phoneType = primaryPhone.get("phoneType") or "" ratingObj = primaryPhone.get("rating") if isinstance(primaryPhone.get("rating"), dict) else {} # Appended Mobile Component (enrichMobile) if "eventsAppendMobileCode" in row: row["eventsAppendMobileCode"] = safe_get(events, "appendMobile", "code") row["appendedMobilePhoneType"] = phoneType row["appendedMobileE164Format"] = phoneObj.get("e164Format") or "" row["appendedMobileIntlFormat"] = phoneObj.get("intlFormat") or "" row["appendedMobileNatlFormat"] = phoneObj.get("natlFormat") or "" row["appendedMobileCountryAlpha2"] = safe_get(phoneObj, "country", "alpha2") or "" row["appendedMobileCountryTitle"] = safe_get(phoneObj, "country", "title") or "" ituCode = phoneObj.get("ituCountryCode") row["appendedMobileItuCountryCode"] = "" if ituCode is None else ituCode ratingScore = ratingObj.get("score") row["appendedMobileRatingScore"] = "" if ratingScore is None else ratingScore numActScore = safe_get(ratingObj, "numberActivity", "score", default=None) row["appendedMobileRatingNumberActivityScore"] = "" if numActScore is None else numActScore row["appendedMobileRatingNumberActivityCode"] = safe_get(ratingObj, "numberActivity", "code") or "" numMatchScore = safe_get(ratingObj, "numberNameMatch", "score", default=None) row["appendedMobileRatingNumberNameMatchScore"] = "" if numMatchScore is None else numMatchScore row["appendedMobileRatingNumberNameMatchCode"] = safe_get(ratingObj, "numberNameMatch", "code") or "" # Verified Mobile Component (verifyMobile) if "eventsVerifyMobileCode" in row: row["eventsVerifyMobileCode"] = safe_get(events, "verifyMobile", "code") row["verifiedMobilePhoneType"] = phoneType row["verifiedMobileE164Format"] = phoneObj.get("e164Format") or "" row["verifiedMobileIntlFormat"] = phoneObj.get("intlFormat") or "" row["verifiedMobileNatlFormat"] = phoneObj.get("natlFormat") or "" row["verifiedMobileIsValid"] = format_bool(phoneObj.get("isValid")) row["verifiedMobileIsPossible"] = format_bool(phoneObj.get("isPossible")) row["verifiedMobileCountryAlpha2"] = safe_get(phoneObj, "country", "alpha2") or "" row["verifiedMobileCountryTitle"] = safe_get(phoneObj, "country", "title") or "" ituCode = phoneObj.get("ituCountryCode") row["verifiedMobileItuCountryCode"] = "" if ituCode is None else ituCode ratingScore = ratingObj.get("score") row["verifiedMobileRatingScore"] = "" if ratingScore is None else ratingScore numActScore = safe_get(ratingObj, "numberActivity", "score", default=None) row["verifiedMobileRatingNumberActivityScore"] = "" if numActScore is None else numActScore row["verifiedMobileRatingNumberActivityCode"] = safe_get(ratingObj, "numberActivity", "code") or "" numMatchScore = safe_get(ratingObj, "numberNameMatch", "score", default=None) row["verifiedMobileRatingNumberNameMatchScore"] = "" if numMatchScore is None else numMatchScore row["verifiedMobileRatingNumberNameMatchCode"] = safe_get(ratingObj, "numberNameMatch", "code") or "" # Verified Phone Component (verifyPhone) if "eventsVerifyPhoneCode" in row: row["eventsVerifyPhoneCode"] = safe_get(events, "verifyPhone", "code") row["verifiedPhonePhoneType"] = phoneType row["verifiedPhoneE164Format"] = phoneObj.get("e164Format") or "" row["verifiedPhoneIntlFormat"] = phoneObj.get("intlFormat") or "" row["verifiedPhoneNatlFormat"] = phoneObj.get("natlFormat") or "" row["verifiedPhoneIsValid"] = format_bool(phoneObj.get("isValid")) row["verifiedPhoneIsPossible"] = format_bool(phoneObj.get("isPossible")) row["verifiedPhoneCountryAlpha2"] = safe_get(phoneObj, "country", "alpha2") or "" row["verifiedPhoneCountryTitle"] = safe_get(phoneObj, "country", "title") or "" ituCode = phoneObj.get("ituCountryCode") row["verifiedPhoneItuCountryCode"] = "" if ituCode is None else ituCode ratingScore = ratingObj.get("score") row["verifiedPhoneRatingScore"] = "" if ratingScore is None else ratingScore numActScore = safe_get(ratingObj, "numberActivity", "score", default=None) row["verifiedPhoneRatingNumberActivityScore"] = "" if numActScore is None else numActScore row["verifiedPhoneRatingNumberActivityCode"] = safe_get(ratingObj, "numberActivity", "code") or "" numMatchScore = safe_get(ratingObj, "numberNameMatch", "score", default=None) row["verifiedPhoneRatingNumberNameMatchScore"] = "" if numMatchScore is None else numMatchScore row["verifiedPhoneRatingNumberNameMatchCode"] = safe_get(ratingObj, "numberNameMatch", "code") or "" def flatten_job_v2_to_row( body: Dict[str, Any], requestId: Optional[str] = None, jobType: Optional[str] = None, ) -> Dict[str, Any]: """Flatten a single JobV2 completed response into a 1:1 tabular CSV row. Deterministic 1:1 mapping adhering to the 'No Arrays' rule by extracting the primary ([0]) index of entity collections. """ effectiveJobType = jobType or body.get("jobType") or "" headers = get_csv_headers(effectiveJobType) # Initialize all columns to empty string for defensive defaulting row: Dict[str, Any] = {header: "" for header in headers} resolvedRequestId = (requestId if requestId and requestId != "unknown" else None) or body.get("id") or "" row["requestId"] = resolvedRequestId row["status"] = body.get("status") or "" events = body.get("events") if isinstance(body.get("events"), dict) else {} payload = body.get("payload") if isinstance(body.get("payload"), dict) else {} # Reason handling (4.4) reasonObj = body.get("reason") if isinstance(reasonObj, dict): row["reasonType"] = reasonObj.get("type") or "" row["reasonTitle"] = reasonObj.get("title") or "" row["reasonBody"] = str(reasonObj.get("body") or reasonObj.get("message") or "") elif isinstance(reasonObj, str): row["reasonBody"] = reasonObj # Correlation tags (4.10, 4.11) tags = payload.get("tags") if isinstance(payload.get("tags"), list) else [] recordId = "" contactId = "" for tag in tags: if isinstance(tag, dict): if tag.get("name") == "recordId" and not recordId: recordId = str(tag.get("value") or "") elif tag.get("name") == "contactId" and not contactId: contactId = str(tag.get("value") or "") if not contactId: contactId = str(safe_get(payload, "person", "source", "objectId") or "") row["recordId"] = recordId row["contactId"] = contactId # Person Name & Validity Component (4.9) personObj = payload.get("person") if isinstance(payload.get("person"), dict) else {} row["firstName"] = personObj.get("firstName") or "" row["middleNames"] = personObj.get("middleNames") or "" row["lastName"] = personObj.get("lastName") or "" row["personIsValid"] = format_bool(personObj.get("isValid")) # Extract primary collections urls = payload.get("urls") if isinstance(payload.get("urls"), list) else [] primaryUrl = urls[0] if (urls and isinstance(urls[0], dict)) else {} urlObj = primaryUrl.get("url") if isinstance(primaryUrl.get("url"), dict) else {} phoneNumbers = payload.get("phoneNumbers") if isinstance(payload.get("phoneNumbers"), list) else [] primaryPhone = phoneNumbers[0] if (phoneNumbers and isinstance(phoneNumbers[0], dict)) else {} # Email Component extraction (4.12) if "validatedEmailNormalFormat" in row: emailAddresses = payload.get("emailAddresses") if isinstance(payload.get("emailAddresses"), list) else [] primaryEmail = emailAddresses[0] if (emailAddresses and isinstance(emailAddresses[0], dict)) else {} emailObj = primaryEmail.get("emailAddress") if isinstance(primaryEmail.get("emailAddress"), dict) else {} row["validatedEmailNormalFormat"] = emailObj.get("normalFormat") or "" row["validatedEmailIsSyntaxValid"] = format_bool(emailObj.get("isSyntaxValid")) row["validatedEmailIsDeliverable"] = format_bool(emailObj.get("isDeliverable")) # Employment Component (enrichMobile, verifyEmployment) if "eventsVerifyEmploymentCode" in row: employmentCode = safe_get(events, "verifyEmployment", "code") row["eventsVerifyEmploymentCode"] = employmentCode # Current employer fallback companyVerificationTitle = safe_get(payload, "employment", "company", "verification", "title") companyTitle = safe_get(payload, "employment", "company", "title") row["verifiedEmploymentCurrentEmployer"] = companyVerificationTitle or companyTitle or "" # Current position fallback positionVerificationTitle = safe_get(payload, "employment", "position", "verification", "title") positionTitle = safe_get(payload, "employment", "position", "title") row["verifiedEmploymentCurrentPosition"] = positionVerificationTitle or positionTitle or "" # Departure handling if employmentCode in ("changed", "verified_elsewhere"): departure = safe_get(payload, "employment", "employmentEnd", "iso8601Format") row["verifiedEmploymentLeftTargetCompany"] = departure if departure else "true" else: row["verifiedEmploymentLeftTargetCompany"] = "" # Locations Component (enrichMobile, verifyEmployment) if "locationsFormattedAddress" in row: locations = payload.get("locations") if isinstance(payload.get("locations"), list) else [] primaryLoc = locations[0] if (locations and isinstance(locations[0], dict)) else {} row["locationsFormattedAddress"] = primaryLoc.get("formattedAddress") or "" # LinkedIn Routing Components (enrichMobile, verifyEmployment, verifyMobile, verifyPhone) if "validatedLinkedInNormalFormat" in row: appendLinkedInCode = safe_get(events, "appendLinkedIn", "code") if appendLinkedInCode in ("from_input", "", None): row["validatedLinkedInNormalFormat"] = safe_get(urlObj, "normalFormat") or "" isSyntaxValid = urlObj.get("isSyntaxValid") row["validatedLinkedInIsSyntaxValid"] = format_bool(isSyntaxValid) isRemote = urlObj.get("isRemote") row["validatedLinkedInIsRemote"] = format_bool(isRemote) rels = primaryUrl.get("rels") if isinstance(rels, list): row["validatedLinkedInRels"] = ",".join(str(r) for r in rels) elif rels is not None: row["validatedLinkedInRels"] = str(rels) else: row["validatedLinkedInRels"] = "" elif appendLinkedInCode == "appended": row["appendedLinkedInNormalFormat"] = safe_get(urlObj, "normalFormat") elif appendLinkedInCode == "replaced": row["replacedLinkedInNormalFormat"] = safe_get(urlObj, "normalFormat") # Phone Component (enrichMobile, verifyMobile, verifyPhone) _flatten_phone_details(row, events, primaryPhone) # Enrich LinkedIn Component (enrichLinkedIn) if "eventsAppendLinkedInCode" in row: appendLinkedInCode = safe_get(events, "appendLinkedIn", "code") row["eventsAppendLinkedInCode"] = appendLinkedInCode if appendLinkedInCode == "appended": row["appendedLinkedInNormalFormat"] = safe_get(urlObj, "normalFormat") return row def write_csv_row(csvPath: str, fieldnames: List[str], rowData: Dict[str, Any]) -> None: """Append a single row to the CSV file, creating directories and header if needed.""" with _csvLock: targetDir = os.path.dirname(csvPath) if targetDir: os.makedirs(targetDir, exist_ok=True) fileExists = os.path.exists(csvPath) and os.path.getsize(csvPath) > 0 with open(csvPath, "a", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") if not fileExists: writer.writeheader() writer.writerow(rowData) f.flush() def _parse_discard_reason( discardJson: Optional[Dict[str, Any]], discardReasonType: str, rawBody: str, ) -> Tuple[str, str, str]: """Extract reason type, title, and body from discarded message json or fallbacks.""" dReasonType = discardReasonType dTitle = "Job Request Discarded" dBody = rawBody[:500] if not isinstance(discardJson, dict): return dReasonType, dTitle, dBody if "reasonType" in discardJson: dReasonType = str(discardJson.get("reasonType") or discardReasonType) if "title" in discardJson: dTitle = str(discardJson.get("title") or dTitle) if "body" in discardJson: dBody = str(discardJson.get("body") or discardJson.get("message") or dBody) elif "message" in discardJson: dBody = str(discardJson.get("message") or dBody) elif "reason" in discardJson and isinstance(discardJson["reason"], dict): r = discardJson["reason"] dReasonType = str(r.get("type") or r.get("reasonType") or dReasonType) dTitle = str(r.get("title") or dTitle) dBody = str(r.get("body") or r.get("message") or dBody) return dReasonType, dTitle, dBody def _populate_discard_correlation( discardJson: Optional[Dict[str, Any]], discardId: str, discardRow: Dict[str, Any], cachePath: str = DEFAULT_CACHE_PATH, ) -> None: """Populate correlation tags and person metadata from request cache or discarded payload.""" cachedData = load_request_cache(cachePath) cached = cachedData.get(discardId) if isinstance(cached, dict): if not discardRow["recordId"] and cached.get("recordId"): discardRow["recordId"] = str(cached["recordId"]) if not discardRow["contactId"] and cached.get("contactId"): discardRow["contactId"] = str(cached["contactId"]) if not discardRow["firstName"] and cached.get("firstName"): discardRow["firstName"] = str(cached["firstName"]) if not discardRow["middleNames"] and (cached.get("middleNames") or cached.get("middleName")): discardRow["middleNames"] = str(cached.get("middleNames") or cached.get("middleName")) if not discardRow["lastName"] and cached.get("lastName"): discardRow["lastName"] = str(cached["lastName"]) if not isinstance(discardJson, dict): return dPayload = discardJson.get("payload") if not isinstance(dPayload, dict): return dTags = dPayload.get("tags") if isinstance(dTags, list): for dTag in dTags: if isinstance(dTag, dict): if dTag.get("name") == "recordId" and not discardRow["recordId"]: discardRow["recordId"] = str(dTag.get("value") or "") elif dTag.get("name") == "contactId" and not discardRow["contactId"]: discardRow["contactId"] = str(dTag.get("value") or "") dPerson = dPayload.get("person") if isinstance(dPerson, dict): if not discardRow["firstName"] and dPerson.get("firstName"): discardRow["firstName"] = dPerson.get("firstName") or "" if not discardRow["middleNames"] and dPerson.get("middleNames"): discardRow["middleNames"] = dPerson.get("middleNames") or "" if not discardRow["lastName"] and dPerson.get("lastName"): discardRow["lastName"] = dPerson.get("lastName") or "" if not discardRow["contactId"]: discardRow["contactId"] = str(safe_get(dPerson, "source", "objectId") or "") def _handle_discarded_message( baseOutputDir: str, jobType: str, rawRequestId: Optional[str], discardReasonType: str, rawBody: str, enableCsv: bool, csvPath: Optional[str], enableJson: bool = True, discardStats: Optional[Dict[Tuple[str, str, str], int]] = None, cachePath: str = DEFAULT_CACHE_PATH, ) -> None: """Handle messages discarded before job creation, saving JSON, appending to CSV, and tracking stats.""" discardId = rawRequestId or "unknown" try: discardJson = json.loads(rawBody) except json.JSONDecodeError: discardJson = None dReasonType, dTitle, dBody = _parse_discard_reason(discardJson, discardReasonType, rawBody) print( f"[discarded] [{jobType}] {discardId} - no job created due to {dReasonType} - {dBody}", flush=True, ) if discardStats is not None: with _seenLock: key = (dReasonType, dTitle, dBody) discardStats[key] = discardStats.get(key, 0) + 1 if enableJson: dPayload: Dict[str, Any] = {} if isinstance(discardJson, dict): rawPayload = discardJson.get("payload") dPayload = rawPayload if isinstance(rawPayload, dict) else discardJson discardBody: Dict[str, Any] = { "id": discardId, "jobType": jobType, "status": "discarded", "reason": { "type": dReasonType, "title": dTitle, "body": dBody, }, "payload": dPayload if dPayload else {"rawBody": rawBody[:500]}, } _write_json_file(baseOutputDir, jobType, discardId, "discarded", discardBody) if not enableCsv: return effectiveCsvPath = csvPath or os.path.join(baseOutputDir, f"{jobType}_results.csv") try: headers = get_csv_headers(jobType) discardRow: Dict[str, Any] = {header: "" for header in headers} discardRow["requestId"] = discardId discardRow["status"] = "discarded" discardRow["reasonType"] = dReasonType discardRow["reasonTitle"] = dTitle discardRow["reasonBody"] = dBody _populate_discard_correlation(discardJson, discardId, discardRow, cachePath=cachePath) # Fallback for recordId if not extracted from payload: infer from requestId (e.g. eval_) if not discardRow["recordId"] and discardId != "unknown" and "_" in discardId: for prefix in ("_rec_", "_record_"): if prefix in discardId: discardRow["recordId"] = discardId[discardId.index(prefix) + 1 :] break else: discardRow["recordId"] = discardId.split("_", 1)[1] write_csv_row(effectiveCsvPath, headers, discardRow) print(f"[discarded] [{jobType}] {discardId} -> {effectiveCsvPath}", flush=True) except Exception as err: print(f"[error] [{jobType}] Failed to append discard CSV {effectiveCsvPath}: {err}", flush=True) def _write_json_file( baseOutputDir: str, jobType: str, requestId: str, status: str, body: Dict[str, Any], ) -> None: """Save raw response JSON body to disk under baseOutputDir/jobType/requestId.json.""" targetDir = os.path.join(baseOutputDir, jobType) os.makedirs(targetDir, exist_ok=True) if not JOB_ID_PATTERN.match(requestId) or requestId in (".", ".."): safeFilename = re.sub(r"[^a-zA-Z0-9\-._]", "_", requestId)[:256] or "unknown" else: safeFilename = requestId filePath = os.path.join(targetDir, f"{safeFilename}.json") try: with open(filePath, "w", encoding="utf-8") as f: json.dump(body, f, ensure_ascii=False, indent=2) print(f"[{status}] [{jobType}] {requestId} -> {filePath}", flush=True) except OSError as err: print(f"[error] [{jobType}] Failed to write {filePath}: {err}", flush=True) def _append_job_to_csv( baseOutputDir: str, jobType: str, csvPath: Optional[str], requestId: str, status: str, body: Dict[str, Any], ) -> None: """Flatten and append a terminal job response to CSV.""" effectiveCsvPath = csvPath or os.path.join(baseOutputDir, f"{jobType}_results.csv") try: headers = get_csv_headers(jobType) rowData = flatten_job_v2_to_row(body, requestId=requestId, jobType=jobType) write_csv_row(effectiveCsvPath, headers, rowData) print(f"[{status}] [{jobType}] {requestId} -> {effectiveCsvPath}", flush=True) except Exception as err: print(f"[error] [{jobType}] Failed to append CSV {effectiveCsvPath}: {err}", flush=True) def handle_response( baseOutputDir: str, jobType: str, message: Any, enableCsv: bool = False, csvPath: Optional[str] = None, enableJson: bool = True, seenRequestIds: Optional[Set[Tuple[str, Optional[str]]]] = None, discardStats: Optional[Dict[Tuple[str, str, str], int]] = None, cachePath: str = DEFAULT_CACHE_PATH, ) -> None: """Process one Pub/Sub response message idempotently. Behavior: - discarded -> log reason, write discarded JSON and CSV row if enabled - terminal statuses (completed, rejected, canceled, failed) -> save JSON body to disk if enabled, log reason, and append to CSV if enabled - ephemeral lifecycle statuses (created, validated, enqueued, etc.) -> log for visibility but do not persist to JSON or append to CSV - invalid JSON / unexpected body type -> log and discard """ rawRequestId = message.attributes.get("modigieJobRequestId") # Check if request was discarded by the service before job creation. discardReasonType = message.attributes.get("modigieJobRequestDiscardReasonType") if discardReasonType: rawBody = message.data.decode("utf-8", errors="replace") _handle_discarded_message( baseOutputDir=baseOutputDir, jobType=jobType, rawRequestId=rawRequestId, discardReasonType=discardReasonType, rawBody=rawBody, enableCsv=enableCsv, csvPath=csvPath, enableJson=enableJson, discardStats=discardStats, cachePath=cachePath, ) return rawBody = message.data.decode("utf-8", errors="replace") try: body = json.loads(rawBody) except json.JSONDecodeError as err: print(f"[unparseable] [{jobType}] {rawRequestId or 'unknown'} - {err}", flush=True) return if not isinstance(body, dict): print( f"[unexpected] [{jobType}] {rawRequestId or 'unknown'} - body is not an object: {type(body).__name__}", flush=True, ) return # Fall back to body["id"] if modigieJobRequestId attribute was missing requestId = rawRequestId or body.get("id") or "unknown" status = body.get("status") or "noStatus" updateTime = body.get("updateTime") if seenRequestIds is not None: with _seenLock: seenKey = (requestId, updateTime) if seenKey in seenRequestIds: print( f"[duplicate] [{jobType}] {requestId} - already processed (updateTime: {updateTime}), skipping", flush=True, ) return seenRequestIds.add(seenKey) if status in TERMINAL_JOB_STATUSES: if enableJson: _write_json_file(baseOutputDir, jobType, requestId, status, body) if enableCsv: _append_job_to_csv(baseOutputDir, jobType, csvPath, requestId, status, body) else: # Intermediate / ephemeral lifecycle state: log only, skip JSON and CSV reason = extract_reason(body) suffix = f" - {reason}" if reason else "" print(f"[{status}] [{jobType}] {requestId}{suffix}", flush=True) def listen_for_responses( outputDir: str, jobType: str, jobConfig: config.UtilConfig, enableCsv: bool = False, csvPath: Optional[str] = None, enableJson: bool = True, seenRequestIds: Optional[Set[Tuple[str, Optional[str]]]] = None, cachePath: str = DEFAULT_CACHE_PATH, ) -> None: """Start a streaming pull subscriber and process messages until interrupted.""" print(WARNING_BANNER) subscriptionPath = jobConfig.get_subscription_path(jobType) creds = jobConfig.get_credentials() subscriberClient = pubsub.SubscriberClient(credentials=creds) resolvedCsvPath = csvPath or os.path.join(outputDir, f"{jobType}_results.csv") targetDir = os.path.join(outputDir, jobType) activeSeenRequestIds = seenRequestIds if seenRequestIds is not None else set() discard_stats: Dict[Tuple[str, str, str], int] = {} print(f"Active Job Type: {jobType}") print(f"Listening on Subscription: {subscriptionPath}") if enableJson: print(f"Completed Responses Dir: {os.path.abspath(targetDir)}") if enableCsv: print(f"Flattened CSV File: {os.path.abspath(resolvedCsvPath)}") csvDir = os.path.dirname(resolvedCsvPath) if csvDir: os.makedirs(csvDir, exist_ok=True) if not os.path.exists(resolvedCsvPath) or os.path.getsize(resolvedCsvPath) == 0: headers = get_csv_headers(jobType) with open(resolvedCsvPath, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=headers) writer.writeheader() print("Press Ctrl-C to stop.\n") def callback(message: Any) -> None: try: handle_response( baseOutputDir=outputDir, jobType=jobType, message=message, enableCsv=enableCsv, csvPath=resolvedCsvPath, enableJson=enableJson, seenRequestIds=activeSeenRequestIds, discardStats=discard_stats, cachePath=cachePath, ) message.ack() except Exception as exc: msgRequestId = message.attributes.get("modigieJobRequestId", "unknown") msgRaw = message.data.decode("utf-8", errors="replace") print(f"Error handling {msgRequestId}: {exc}", flush=True) print(f" Raw body: {msgRaw[:500]}", flush=True) # Ack to avoid blocking the subscription backlog message.ack() streamingPullFuture = subscriberClient.subscribe(subscriptionPath, callback=callback) with subscriberClient: try: streamingPullFuture.result() except exceptions.PermissionDenied as exc: print(f"\n[Permission Denied: 403] {exc.message or exc}") print("\nTip: Subscribing requires 'roles/pubsub.subscriber' and 'roles/pubsub.viewer'") print("on the pull subscription in the 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") except (KeyboardInterrupt, FuturesTimeoutError): streamingPullFuture.cancel() try: streamingPullFuture.result() except Exception as cancelErr: _ = cancelErr print("\nStopped.") if discard_stats: print("\n" + "=" * 80) print("DISCARDED JOB REQUESTS SUMMARY") print("=" * 80) print(f"{'COUNT':<7} | {'REASON TYPE':<30} | {'TITLE':<30}") print("-" * 80) for (rtype, rtitle, rbody), count in sorted(discard_stats.items(), key=lambda x: x[1], reverse=True): print(f"{count:<7} | {rtype:<30} | {rtitle:<30}") print(f" > {rbody}") print("=" * 80 + "\n") def main() -> None: """CLI entrypoint for subscriber.""" parser = argparse.ArgumentParser( description="Subscribe to Modigie Cloud Pub/Sub job responses and write completed ones to disk." ) parser.add_argument( "output_dir_pos", nargs="?", default=None, help="Optional positional destination directory for response JSON files.", ) parser.add_argument( "--outputDir", dest="output_dir_flag", default=None, help=f"Destination directory for completed response JSON files (default: ./{DEFAULT_OUTPUT_DIR}).", ) parser.add_argument( "--jobType", default=None, choices=list(config.SUPPORTED_JOB_TYPES), help=f"Target Modigie job type to listen for. 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( "--credentials", "--keyFile", dest="credentialsPath", default=None, help="Optional path to Google Cloud service account JSON key file (overrides MODIGIE_CREDENTIALS_JSON).", ) parser.add_argument( "--csv", action=argparse.BooleanOptionalAction, default=None, help=( "Save completed responses to a flattened CSV file " "(default: enabled if --outputCsv is provided, otherwise False)." ), ) parser.add_argument( "--outputCsv", "--output-csv", dest="outputCsv", default=None, help="Custom destination file path for the flattened CSV (default: /_results.csv).", ) parser.add_argument( "--json", action=argparse.BooleanOptionalAction, default=True, help="Save individual raw JSON files (default: True; use --no-json to disable).", ) args = parser.parse_args() activeJobType = args.jobType or DEFAULT_JOB_TYPE outputDir = args.output_dir_flag or args.output_dir_pos or DEFAULT_OUTPUT_DIR outputCsv = args.outputCsv if args.csv is not None: enableCsv = args.csv else: enableCsv = bool(outputCsv) enableJson = bool(args.json) jobConfig = config.UtilConfig( tenantProjectId=args.tenantProjectId, repoShortId=args.repoShortId, credentialsPath=args.credentialsPath, ) listen_for_responses( outputDir=outputDir, jobType=activeJobType, jobConfig=jobConfig, enableCsv=enableCsv, csvPath=outputCsv, enableJson=enableJson, ) if __name__ == "__main__": main()