-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathcreate_cis.py
128 lines (111 loc) · 6.24 KB
/
create_cis.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""
Program to create GDSN Catalogue Item Subscription (CIS) messages based on a list of supplier GLNS.
Author: Sjoerd Schaper - GS1 Nederland
"""
import csv
import sys
import time
import os
import random
import string
from pathlib import Path
def get_random_string(length):
letters = string.ascii_uppercase
result_str = ''.join(random.choice(letters) for i in range(length))
return result_str
if len(sys.argv) > 1:
source_file = sys.argv[1]
else:
source_file = "test_file"
# The input file has all possible parameters in it.
# see README for a explanation
headers = ['gln_dr','gln_ds','gpc','gtin', 'tm']
infile = Path(f'./input/{source_file}.csv')
# Change this to the GLN of your data pool
data_pool_gln = '8712345013042'
cntr = 1
b_nr = 1
# To prevent overloading the data pool the messages are split up in batches
batch_size = 100
batch = os.path.join('output',source_file,'batch_001')
if not os.path.exists(batch):
os.makedirs(batch)
with open(infile, 'r', encoding='utf-8', errors='ignore') as fp:
reader = csv.DictReader(fp, fieldnames=headers, skipinitialspace=True)
for row in reader:
if row.get('gln_dr') != 'gln_dr':
print(f"gln_ds: {row.get('gln_ds')}, gpc: {row.get('gpc')}, gtin: {row.get('gtin')}, tm: {row.get('tm')}")
time_sys = time.strftime("%Y-%m-%dT%H:%M:%S")
timestr = time.strftime("%Y_%m_%dT%H_%M_%S")
file_id = get_random_string(8)
inst_id = get_random_string(8)
if cntr % batch_size == 0:
b_nr = b_nr + 1
batch = os.path.join('output',source_file, 'batch_' + str(b_nr).zfill(3))
if not os.path.exists(batch):
os.makedirs(batch)
file_name = os.path.join(batch, f'CIS_{source_file.upper()}_{row.get("gln_dr")}_{row.get("gln_ds")}_{file_id}.xml')
outfile = open(str(file_name), "w", encoding='utf-8')
outfile.write('<?xml version="1.0" encoding="utf-8"?>\n')
outfile.write('<catalogue_item_subscription:catalogueItemSubscriptionMessage xmlns:catalogue_item_subscription="urn:gs1:gdsn:catalogue_item_subscription:xsd:3" xmlns:sh="http://www.unece.org/cefact/namespaces/StandardBusinessDocumentHeader" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="urn:gs1:gdsn:catalogue_item_subscription:xsd:3 http://www.gdsregistry.org/3.1/schemas/gs1/gdsn/CatalogueItemSubscription.xsd">\n')
outfile.write('<sh:StandardBusinessDocumentHeader>\n')
outfile.write('<sh:HeaderVersion>1.0</sh:HeaderVersion>\n')
outfile.write('<sh:Sender>\n')
outfile.write(f'<sh:Identifier Authority="GS1">{row.get("gln_dr")}</sh:Identifier>\n')
outfile.write('</sh:Sender>\n')
outfile.write('<sh:Receiver>\n')
outfile.write(f'<sh:Identifier Authority="GS1">{data_pool_gln}</sh:Identifier>\n')
outfile.write('</sh:Receiver>\n')
outfile.write('<sh:DocumentIdentification>\n')
outfile.write('<sh:Standard>GS1</sh:Standard>\n')
outfile.write('<sh:TypeVersion>3.1</sh:TypeVersion>\n')
outfile.write(f'<sh:InstanceIdentifier>CIS_{row.get("gln_dr")}_{timestr}_{inst_id}</sh:InstanceIdentifier>\n')
outfile.write('<sh:Type>catalogueItemSubscription</sh:Type>\n')
outfile.write(f'<sh:CreationDateAndTime>{time_sys}</sh:CreationDateAndTime>\n')
outfile.write('</sh:DocumentIdentification>\n')
outfile.write('</sh:StandardBusinessDocumentHeader>\n')
# transactiom
trans_id = get_random_string(8)
outfile.write('<transaction>\n')
outfile.write('<transactionIdentification>\n')
outfile.write(f'<entityIdentification>CIS_ADD_{row.get("gln_dr")}_{trans_id}</entityIdentification>\n')
outfile.write('<contentOwner>\n')
outfile.write(f'<gln>{row.get("gln_dr")}</gln>\n')
outfile.write('</contentOwner>\n')
outfile.write('</transactionIdentification>\n')
# document
doc_id = get_random_string(8)
outfile.write('<documentCommand>\n')
outfile.write('<documentCommandHeader type="ADD">\n')
outfile.write('<documentCommandIdentification>\n')
outfile.write(f'<entityIdentification>CIS_ADD_{row.get("gln_dr")}_{doc_id}</entityIdentification>\n')
outfile.write('<contentOwner>\n')
outfile.write(f'<gln>{row.get("gln_dr")}</gln>\n')
outfile.write('</contentOwner>\n')
outfile.write('</documentCommandIdentification>\n')
outfile.write('</documentCommandHeader>\n')
outfile.write('<catalogue_item_subscription:catalogueItemSubscription>\n')
outfile.write(f'<creationDateTime>{time_sys}</creationDateTime>\n')
outfile.write('<documentStatusCode>ORIGINAL</documentStatusCode>\n')
outfile.write('<catalogueItemSubscriptionIdentification>\n')
outfile.write(f'<entityIdentification>Subscription_{row.get("gln_ds")}_{row.get("tm")}_{doc_id}</entityIdentification>\n')
outfile.write('<contentOwner>\n')
outfile.write(f'<gln>{row.get("gln_dr")}</gln>\n')
outfile.write('</contentOwner>\n')
outfile.write('</catalogueItemSubscriptionIdentification>\n')
outfile.write(f'<dataRecipient>{row.get("gln_dr")}</dataRecipient>\n')
if row.get("gln_ds") != '0':
outfile.write(f'<dataSource>{row.get("gln_ds")}</dataSource>\n')
if row.get("gpc") != '0':
outfile.write(f'<gpcCategoryCode>{row.get("gpc")}</gpcCategoryCode>\n')
if row.get("gtin") != '0':
outfile.write(f'<gtin>{row.get("gtin")}</gtin>\n')
outfile.write('<targetMarket>\n')
outfile.write(f'<targetMarketCountryCode>{row.get("tm")}</targetMarketCountryCode>\n')
outfile.write('</targetMarket>\n')
outfile.write('</catalogue_item_subscription:catalogueItemSubscription>\n')
outfile.write('</documentCommand>\n')
outfile.write('</transaction>\n')
cntr = cntr + 1
outfile.write('</catalogue_item_subscription:catalogueItemSubscriptionMessage>\n')
outfile.close()