Python script to upload the .csv file:
[Side note1: Here, I upload several different .csv files, to several different Kobo forms, which adds a tad of complexity]
[Side note2: If someone would be kind enough to explain me how to properly insert code, I would be very happy]
#this work is based of this forum post:
#Using API to upload media files? - #8 by Josh
import requests, json
from datetime import datetime
import os
string_output = “”
#This uploads a media file whose path is given by FILENAME (if there is already a file with the same name, deletes it
#to the Kobo form whose asset_uid is XFORM
#If REDEPLOY (boolean) is true, then also redeploys the form on Kobo
def upload(XFORM,FILENAME,REDEPLOY):
URL = ‘KoboToolbox Form Building API’
TOKEN = ‘YOUR TOKEN’
headers = {‘Authorization’: f’Token {TOKEN}'}
global string_output
#### TO DELETE OLD MEDIA FILE
try:
data_url = "%sassets/%s/files/" % (URL, XFORM)
response = requests.get(data_url, headers=headers, params={'format': 'json'})
json_uid = response.json()
#Look throught the different media files attached to the form, and look for the one whose filename matches the title of the file we want to upload
for result in json_uid['results']:
#print(str(result['metadata']['filename']) + " vs " + str(FILENAME.split("\\")[-1]))
if result['metadata']['filename'] == FILENAME.split("\\")[-1]:
media_uid = result['uid']
asset_url = "%s%s/" % (data_url, media_uid)
#If we have found a file, we will delete it. If not an error will be triggered and caught
res = requests.delete(asset_url, headers=headers)
print(res)
print("Old media file deleted successfully")
string_output += "Old media file deleted successfully" + "\n"
except:
print("No media file deleted")
string_output += "No media file deleted" + "\n"
pass
#### TO UPLOAD NEW MEDIA FILE
try:
KC_URL= "%sassets/%s/files.json" % (URL, XFORM)
MIME = 'text/csv'
payload = {'filename': FILENAME}
files = {'content': open(FILENAME, 'rb')}
data = {'description': 'Input and equipment media file', 'metadata': json.dumps(payload), 'file_type': 'form_media'}
res = requests.post(url=KC_URL, headers=headers,data=data, files=files)
print(res)
print("New media file uploaded successfully")
string_output +="New media file uploaded successfully" + "\n"
except:
print("No media file")
string_output += "No media file" + "\n"
#### TO REDEPLOY FORM
if REDEPLOY:
asset_url = "%sassets/%s/" % (URL, XFORM)
headers = {
'Accept': 'application/json',
'Authorization': f'Token {TOKEN}'
}
response = requests.get(asset_url, headers=headers, params={'format': 'json'})
version_to_deploy = response.json()['version_id']
deployment_data = {
'version_id': version_to_deploy,
'active': True
}
response = requests.patch(asset_url + 'deployment/', headers=headers, data=deployment_data)
print(response)
print("Form redeployed successfully")
string_output += "Form redeployed successfully" + "\n"
#----------------------------------
#List of Kobo form asset_uid
[Put the here the ID of the form(s) to which the .csv file(s) should be uploaded]
FORM_1_ID = “”
FORM_2_ID = “”
#etc.
#Path to the different csv files to be uploaded
GENERAL_PATH = r"G:\Shared drives\Drive\Folder"
CSV_PATH = r"\CSV subfolder"
CSV_FILE_1_PATH = GENERAL_PATH + CSV_PATH + r"\csv_file1.csv"
CSV_FILE_2_PATH = GENERAL_PATH + CSV_PATH + r"\csv_file2.csv"
#etc.
print("project path: " + PROJECT_CSV_PATH)
#Linking the forms to the relevant csv
INPUT_LIST = {
“form1”: {“form_id”:FORM_1_ID,“csvs”:[CSV_FILE_1_PATH,CSV_FILE_3_PATH]},
“form2”: {“form_id”:FORM_2_ID,“csvs”:[CSV_FILE_2_PATH,CSV_FILE_3_PATH]},
#etc.
}
print(INPUT_LIST[‘user’][‘csvs’][0])
print(INPUT_LIST[‘user’][‘csvs’][0].split(“\”)[-1])
print(“\”)
note that we upload the .csv files one by one.
#It would surely be more efficient to upload once, for a given form, but I could not justify the time to do it!
for form in INPUT_LIST:
print ("updating " + form)
string_output += "updating " + form + “\n”
for csv in INPUT_LIST[form][“csvs”]:
print ("uploading " + csv)
string_output += "uploading " + csv + “\n”
if csv == INPUT_LIST[form][“csvs”][-1]:
upload(INPUT_LIST[form][“form_id”],csv,True)
else:
upload(INPUT_LIST[form][“form_id”],csv,False)
string_output += “\n”
#Write a little summary of the update in a newly created file
OUTPUT_FILE_PATH = GENERAL_PATH + r"some path "
complete_name = OUTPUT_FILE_PATH + datetime.now().strftime(“%d-%m-%Y %H-%M-%S”)+ “.txt”
file1 = open(complete_name,“w”)
file1 = open(complete_name,“w”)
file1.write(string_output)
file1.close()
