import csv import os from pathlib import Path import shutil import zipfile from owsclient import M2MTokenManager, OwsClient import phpserialize import config class FetchXmlError(Exception): pass m2m_token_manager = M2MTokenManager( environment=config.TOKEN_MANAGER_ENVIRONMENT, service_name=config.SERVICE_NAME, secrets_manager=config.secrets_manager_client, ) client = OwsClient( environment=config.ENVIRONMENT, service_name=config.SERVICE_NAME, timeout=config.DEFAULT_OWS_CLIENT_TIMEOUT, m2m_token_manager=m2m_token_manager, ) def main(): # Use environment variables with no default values paramMobile = os.environ["RINGTONE"] paramEnvironment = os.environ["ENVIRONMENT"] paramType = os.environ["TYPE"] input_path = Path("./input/input.csv") output_dir = Path("./output") output_dir.mkdir(parents=True, exist_ok=True) errors_dir = output_dir / "errors" delete_contents(output_dir) if not input_path.exists(): write_job_file( os.environ["DMS_ID"], os.environ["UPC"], os.environ["FORMAT"], paramType, paramMobile, paramEnvironment, output_dir, errors_dir, True, ) else: with input_path.open(newline="") as f: reader = csv.DictReader(f) if ( not reader.fieldnames or "DMS_ID" not in reader.fieldnames or "UPC" not in reader.fieldnames or "FORMAT" not in reader.fieldnames ): raise ValueError( "input.csv must have header columns: DMS_ID, UPC, FORMAT" ) row_count = 0 for row in reader: write_job_file( row["DMS_ID"], row["UPC"], row["FORMAT"], paramType, paramMobile, paramEnvironment, output_dir / row["DMS_ID"] / paramType, errors_dir, ) row_count += 1 print(f"Processed {row_count} rows from {input_path}") if row_count > 0: zip_results(output_dir) def delete_contents(target_dir): """Delete all files and directories inside target_dir.""" if not target_dir.exists(): return for child in target_dir.iterdir(): if child.is_dir(): shutil.rmtree(child) else: child.unlink() def construct_url( paramDmsId, paramUPC, paramMobile, paramType, paramFormat, paramEnvironment ): formatKey = paramFormat.replace("[UNDER DEVELOPMENT]", "").lstrip() mobileLink = f"/mobile/1/clip_code/{paramUPC}_1_1_1" if paramMobile == "Yes" else "" formatMap = { "DDEX v. 4.3": { "service": "ows-delivery-metadata", "url": f"products/{paramUPC}/stores/{paramDmsId}/delivery_metadata?format=ddex_ern&version=4.3&delivery_type={paramType}", }, "DDEX v. 3.8": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/storename/ddexern/version/3.8.0/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "DDEX v. 3.7": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/storename/ddexern/version/3.7.0/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "DDEX v. 3.4": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/storename/ddexern/version/3.4.0/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "DDEX v. 3.2": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/storename/ddexern/version/1.32.0/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "DDEX RDR 1.4": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/storename/ddexrdr/version/1.4/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "iTunes Music": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/subformat/music/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "iTunes Film": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/subformat/film/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "Orchard 1.5": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/storename/orchard/version/1.5/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "Custom XML (Frmwk 2)": { "service": "ows-metadata", "url": f"oa/metadata/wg_metadata.php/dms_id/{paramDmsId}/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "Custom XML (Frmwk 3)": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "Custom Film XML (Frmwk 3)": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/subformat/film/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, "Custom TV XML (Frmwk 3)": { "service": "ows-metadata", "url": f"oa/metadata/generate/dms_id/{paramDmsId}/subformat/tv/returnErr/1/xml_only/0/delivery_type/{paramType}/upc/{paramUPC}{mobileLink}", }, } hostnameMap = { "ows-metadata": { "qa": "qa-ows-metadata.theorchard.io", "prod": "prod-ows-metadata.theorchard.io", }, "ows-delivery-metadata": { "qa": "qa-ows-delivery-metadata.theorchard.io", "prod": "prod-ows-delivery-metadata.theorchard.io", }, } formatChoice = formatMap[formatKey] hostname = hostnameMap[formatChoice["service"]][paramEnvironment] url = formatChoice["url"] return hostname, formatChoice["service"], url def fetch_xml(dms_id, upc, ringtone, job_type, job_format, environment): hostname, service_name, url = construct_url( dms_id, upc, ringtone, job_type, job_format, environment ) print(f"Querying URL: https://{hostname}/{url}") response = client.get(service_name, url) if response.status_code != 200: raise FetchXmlError( f"HTTP {response.status_code} from {service_name}: {response.text}" ) try: body = phpserialize.loads(response.text.encode("utf-8"), decode_strings=True) decoded_body = body["metadata"] except ValueError: decoded_body = response.text except KeyError: raise FetchXmlError( f"'metadata' key missing in response from {service_name}. " f"Keys present: {list(body.keys())}\n\nBody: {response.text}" ) except Exception as e: raise FetchXmlError( f"Failed to deserialize response from {service_name}: {e}\n\nBody: {response.text}" ) from e return decoded_body def write_job_file( dms_id, upc, job_format, job_type, ringtone, environment, output_dir, errors_dir, print_result=False, ): print(f"Processing DMS_ID={dms_id} UPC={upc} FORMAT={job_format}") try: xml_content = fetch_xml( dms_id, upc, ringtone, job_type, job_format, environment ) except FetchXmlError as e: errors_dir.mkdir(parents=True, exist_ok=True) error_file = errors_dir / f"{dms_id}_{job_type}_{upc}.txt" error_file.write_text(f"{e}\n") print(f"EXCEPTION — written to {error_file}") return out_file = output_dir / f"{upc}.xml" out_file.parent.mkdir(parents=True, exist_ok=True) out_file.write_text(xml_content) if print_result: print(xml_content) def zip_results(output_dir): zip_path = output_dir / "results.zip" with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for file in output_dir.rglob("*"): if file == zip_path or not file.is_file(): continue zf.write(file, file.relative_to(output_dir)) print(f"Zip written to {zip_path}") if __name__ == "__main__": main()