phowell
/
canvasapp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
canvasapp/localcache2.py

244 lines
7.8 KiB
Python
Raw Blame History

# 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
def users_new_this_semester(sem=''):
if not len(sem):
sem = input("which semester? (ex: 202150) ")
users_to_enroll = set()
where1 = "c.sis_source_id LIKE '%s-%%'" % sem
where2 = "c.sis_source_id NOT LIKE '%s-%%'" % sem
q = """SELECT u.id, u.name, u.sortable_name, string_agg(c.course_code, ','), COUNT(e.id) AS num FROM canvas.enrollments AS e
JOIN canvas.users AS u ON e.user_id=u.id
JOIN canvas.courses AS c ON e.course_id=c.id
WHERE %s
AND e.workflow_state='active'
AND e.type='StudentEnrollment'
AND u.id NOT IN (
SELECT u.id FROM canvas.enrollments AS e
JOIN canvas.users AS u ON e.user_id=u.id
JOIN canvas.courses AS c ON e.course_id=c.id
WHERE %s
AND e.workflow_state='active'
AND e.type='StudentEnrollment'
GROUP BY u.id
)
GROUP BY u.id
ORDER BY num DESC, u.sortable_name""" % (where1,where2)
(connection,cursor) = db()
cursor.execute(q)
#s = cursor.fetchall()
#if s:
for u in cursor:
users_to_enroll.add(str(u[0]))
print(u)
print("%i new users this semester." % len(users_to_enroll))
print(users_to_enroll)
return users_to_enroll
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],
4: ['new students this semester', users_new_this_semester],
}
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]()
Reference in New Issue View Git Blame Copy Permalink