# Local data, saving and manipulating import os, re, gzip, codecs, funcy, pytz, json, random, functools, requests, sys, csv, time, psycopg2 import pandas as pd import numpy as np from collections import defaultdict from datetime import datetime as dt from datetime import timedelta from dateutil.parser import parse from os.path import exists, getmtime from pipelines import sync_non_interactive, url, header, gp, dean from tabulate import tabulate ######### ######### LOCAL DB ######### CON = '' CURSOR = '' def db(): global CON,CURSOR CON = psycopg2.connect(database="db", host="192.168.1.6", user="postgres", password="rolley34", port="5432") CURSOR = CON.cursor() return CON,CURSOR ''' # Help the next function to upload new users directly to conf database on gavilan. def employees_refresh_flex(data): try: data['a'] = 'set/newuser' data['sis_user_id'] = data['sis_user_id'][3:] print("\nUploading this: \n") print(json.dumps(data, indent=2)) print("\n") a = input("Continue (y) or skip (n) ? ") if a == 'y': # This is what I was missing.......... # req.add_header("Content-type", "application/x-www-form-urlencoded") r3 = requests.post('https://www.gavilan.edu/staff/flex/2020/api.php', params=data) print(r3.text) #print(r3.headers) except Exception as ex: print("Failed on: %s\nErr: %s" % (str(data),str(ex))) # Everyone in iLearn DB with an xyz@gavilan.edu email address. def all_gav_employees(): (connection,cursor) = db() connection.row_factory = dict_factory q = """SELECT u.canvasid, u.name, u.created, u.sortablename, h.address, h.type, h.workflow_state, h.updated_at, p.last_request_at, p.last_login_at, p.current_login_at, p.last_login_ip, p.current_login_ip, p.sis_user_id, p.unique_name FROM users AS u JOIN comm_channel AS h ON u.id=h.user_id JOIN pseudonym AS p ON p.user_id=u.id WHERE h.address LIKE "%@gavilan.edu" ORDER BY u.sortablename""" cursor = connection.cursor() cursor.execute(q) everyone = cursor.fetchall() everyone_set = set() for E in everyone: try: everyone_set.add( E['address'].lower() ) except Exception as e: print("Exception: %s\nwith: %s" % (str(e), str(E))) oo = open('cache/temp1.txt','w') oo.write(json.dumps(list(everyone_set), indent=2)) existing = requests.get('https://gavilan.edu/staff/flex/2020/api.php?a=get/users') ex = json.loads( existing.text ) already_enrolled = set() for usr in ex['users']: try: #already_enrolled.add( (usr['goo'], usr['email'].lower(), usr['name']) ) already_enrolled.add( usr['email'].lower() ) except Exception as e: print("Exception: %s\nWith: %s" % (str(e),str(usr))) oo.write( "\n"*20 + '------------------------------------------\n'*20 + '------ - - - - - - ' ) oo.write(json.dumps(list(already_enrolled), indent=2)) # conf_users wants: goo, email, name, active # and emails have random capitalization # name is First Last, and sometimes with Middle in there. # # using sets: to_enroll = [ x for x in students if x not in already_enrolled ] new_emp = [ x for x in everyone_set if x not in already_enrolled ] # take the all_employee list, filter -> anyone who's in 'existing' is removed # funcy.where( lambda x: x['email'] == ae[4] , existing ) #new_emp = list(funcy.filter( lambda ae: funcy.where( existing, email=ae['email'] ), all_emp )) #new_emp = list(funcy.where( existing, email=b'phowell@gavilan.edu')) #ae['email'] )) print(new_emp) oo.write( "\n"*20 + '------------------------------------------\n'*20 + '------ - - - - - - ' ) oo.write(json.dumps(list(new_emp), indent=2)) # Now, iLearn db (everyone)... find the rows that match the email addresses # that we've decided we need to add (new_emp) #print(everyone) #print( "searching for %s" % j ) #print( "searched for %s, found: %s" % (j, str(to_add) )) #print("\nUploading...\n") for j in new_emp: #j = new_emp[0] print(j) to_add = list(funcy.where( everyone, address=j )) if to_add: employees_refresh_flex(to_add[0]) else: print("Didn't find an entry for that account.") print("done uploading") ''' def teachers_by_term(TERM = "202430"): q = f"""SELECT c.id, c.name, c.course_code, c.sis_source_id, c.created_at, c.start_at, c.workflow_state, e.last_attended_at, u.id, u.sortable_name, u.created_at FROM canvas.courses AS c JOIN canvas.enrollments AS e ON e.course_id=c.id JOIN canvas.users AS u ON u.id=e.user_id WHERE c.sis_source_id LIKE '{TERM}%' AND e.type='TeacherEnrollment' ORDER BY u.sortable_name, c.course_code;""" (connection,cursor) = db() cursor.execute(q) all_teachers = cursor.fetchall() table = [ [t[9],t[1],t[3],t[6]] for t in all_teachers] print(tabulate(table)) #for t in all_teachers: # print("\t".join( [str(x) for x in [t[9],t[1],t[3],t[6]]])) return all_teachers def courses_in_term(TERM = "202430"): q = f"""SELECT c.id, c.name, c.course_code, c.sis_source_id, c.workflow_state FROM canvas.courses AS c WHERE c.sis_source_id LIKE '{TERM}%' ORDER BY c.course_code;""" (connection,cursor) = db() cursor.execute(q) all = cursor.fetchall() #table = [ [t[9],t[1],t[3],t[6]] for t in all_teachers] print(tabulate(all)) def pages_in_term(TERM="202430"): # q = f"""SELECT c.id, c.course_code, c.sis_source_id, wp.id as wp_id, wp.title, wp.url, c.name , wp.body FROM canvas.courses c JOIN canvas.wiki_pages wp ON wp.context_id=c.id WHERE c.sis_source_id LIKE '{TERM}%' ORDER BY c.sis_source_id, wp.title;""" (connection,cursor) = db() cursor.execute(q) all = cursor.fetchall() #print(tabulate(all)) return all if __name__ == "__main__": print ('') options = { 1: ['all teachers', teachers_by_term], 2: ['courses in term', courses_in_term], 3: ['pages in term', pages_in_term] } if len(sys.argv) > 1 and re.search(r'^\d+',sys.argv[1]): resp = int(sys.argv[1]) print("\n\nPerforming: %s\n\n" % options[resp][0]) else: print ('') for key in options: print(str(key) + '.\t' + options[key][0]) print('') resp = input('Choose: ') # Call the function in the options dict options[ int(resp)][1]()