import os import uuid import re import aiofiles import asyncio import logging from fastapi import APIRouter, Depends, Request, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.orm import Session from dependencies.helper import * import db_config.database as dbase from db_config.models import Courses, User, AccessToken, LMSPlatform, Deployment, LMSCourseLog from typing import Dict import httpx from datetime import datetime, timezone, timedelta from fastapi.encoders import jsonable_encoder from fastapi.responses import JSONResponse, FileResponse, RedirectResponse from dotenv import load_dotenv from starlette.config import Config from authlib.integrations.starlette_client import OAuth, OAuthError import json, base64 router = APIRouter() load_dotenv() bearer = HTTPBearer() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) config = Config(".env") # store CLIENT_ID, CLIENT_SECRET, CANVAS_BASE_URL # oauth = OAuth(config) # oauth.register( # name="canvas", # client_id=os.getenv("CANVAS_CLIENT_ID"), # client_secret=os.getenv("CANVAS_CLIENT_SECRET"), # access_token_url=f"{config('CANVAS_BASE_URL')}/login/oauth2/token", # authorize_url=f"{config('CANVAS_BASE_URL')}/login/oauth2/auth", # client_kwargs={ # "scope": # ( # "url:GET|/api/v1/users/:user_id/tokens/:id " # "url:GET|/api/v1/accounts/:account_id/developer_keys " # "url:GET|/api/v1/course_creation_accounts " # "url:GET|/api/v1/courses " # "url:GET|/api/v1/courses/:id " # "url:GET|/api/v1/accounts/:account_id/courses/:id " # "url:GET|/api/v1/users/:user_id/courses " # "url:POST|/api/v1/accounts/:account_id/courses " # "url:POST|/api/v1/courses/:course_id/files " # "url:GET|/api/v1/accounts " # "url:POST|/api/v1/courses/:course_id/content_migrations " # "url:POST|/api/v1/users/:user_id/content_migrations " # "url:GET|/api/v1/accounts/:account_id/content_migrations/:id " # "url:GET|/api/v1/courses/:course_id/content_migrations/:id" # ) # } # your scopes # ) def get_oauth_client(platform: LMSPlatform): oauth = OAuth() base_url = platform.tenant_domain.rstrip("/") oauth.register( name="canvas_dynamic", client_id=platform.rest_client_id if config("ENABLE_GLOBAL_KEY") == "False" else config("CANVAS_CLIENT_ID"), client_secret=platform.rest_client_secret if config("ENABLE_GLOBAL_KEY") == "False" else config("CANVAS_CLIENT_SECRET"), access_token_url=f"{base_url}/login/oauth2/token", authorize_url=f"{base_url}/login/oauth2/auth", client_kwargs={ "scope": ( "url:GET|/api/v1/course_creation_accounts " "url:GET|/api/v1/courses " "url:GET|/api/v1/courses/:id " "url:GET|/api/v1/users/:user_id/courses " "url:GET|/api/v1/users/:id " "url:POST|/api/v1/accounts/:account_id/courses " "url:POST|/api/v1/courses/:course_id/files " "url:POST|/api/v1/courses/:course_id/content_migrations " "url:GET|/api/v1/courses/:course_id/content_migrations/:id" ) } ) return oauth @router.get("/login") async def login( request: Request, tenant_domain: str | None = None, user_id: int | None = None, lms_platform_id: int | None = None, ): try: db = dbase.SessionLocal() ##To use Canvas OAuth dynamically based on platform info from DB platform = db.query(LMSPlatform).filter(LMSPlatform.id == lms_platform_id).first() if not platform: return JSONResponse(status_code=404, content={"error": "Invalid LMS platform ID"}) # Create dynamic OAuth client oauth = get_oauth_client(platform) redirect_uri = request.url_for("auth") # pack custom params into state (JSON -> urlsafe base64) payload = {"tenant_domain": tenant_domain, "user_id": user_id, "lms_platform_id": lms_platform_id} state = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode() logger.info("Initiating login for tenant_domain=%s, user_id=%s, lms_platform_id=%s", tenant_domain, user_id, lms_platform_id) logger.info("Redirect URI: %s", redirect_uri) logger.info("State: %s", state) return await oauth.canvas_dynamic.authorize_redirect(request, redirect_uri, state=state) except Exception as e: return JSONResponse(status_code=500, content={"error": str(e)}) @router.get("/auth") async def auth(request: Request): print("In auth callback") # unpack custom params from state state_b64 = request.query_params.get("state") if not state_b64: return JSONResponse(status_code=400, content={"error": "Missing state parameter"}) try: state_json = base64.urlsafe_b64decode(state_b64.encode()).decode() state = json.loads(state_json) except Exception as e: return JSONResponse(status_code=400, content={"error": "Invalid state parameter", "details": str(e)}) tenant_domain = state.get("tenant_domain") user_id = state.get("user_id") lms_platform_id = state.get("lms_platform_id") logger.info("State: %s", state) logger.info("Tenant Domain: %s, User ID: %s, LMS Platform ID: %s", tenant_domain, user_id, lms_platform_id) logger.info("B64 State: %s", state_b64) db = dbase.SessionLocal() ##To use Canvas OAuth dynamically based on platform info from DB platform = db.query(LMSPlatform).filter(LMSPlatform.id == lms_platform_id).first() oauth = get_oauth_client(platform) ## try: token = await oauth.canvas_dynamic.authorize_access_token(request) except Exception as e: return JSONResponse(status_code=400, content={"error": "Authorization faileds", "details": str(e)}) # response includes access_token, refresh_token, expires_in, canvas_region, user info request.session["token"] = token accounts = await get_course_creation_accounts(access_token=token["access_token"], tenant_domain = platform.tenant_domain) # print("Accounts:", accounts) plateform = db.query(LMSPlatform).filter(LMSPlatform.id == int(lms_platform_id)).first() if plateform: plateform.accounts = json.dumps(accounts) db.commit() account_id = accounts[0]["id"] if accounts else None # developer_keys = await get_developer_keys(access_token=token["access_token"], tenant_domain = platform.tenant_domain, account_id=account_id) if account_id else [] # print("Developer Keys:", developer_keys) request.session["account_id"] = account_id access_token = db.query(AccessToken).filter(AccessToken.user_id == int(user_id)).order_by(AccessToken.id.desc()).first() deployment = db.query(Deployment).filter(Deployment.platform_id == int(lms_platform_id)).order_by(Deployment.id.desc()).first() if not deployment: deployment_code = str(uuid.uuid4()) new_deployment = Deployment( platform_id=int(lms_platform_id), deployment_id=deployment_code ) db.add(new_deployment) db.commit() deployment_id = new_deployment.id else: deployment_id = deployment.id if not access_token: new_token_entry = AccessToken( platform_id=int(lms_platform_id), deployment_id=int(deployment_id), user_id=int(user_id), token=token["access_token"], rest_api_token=token["access_token"], refresh_token=token.get("refresh_token"), expires_at=datetime.utcnow() + timedelta(seconds=token.get("expires_in", 3600)) ) db.add(new_token_entry) db.commit() else: access_token.rest_api_token = token["access_token"] access_token.refresh_token = token.get("refresh_token") access_token.expires_at = datetime.utcnow() + timedelta(seconds=token.get("expires_in", 3600)) db.commit() redirect_url = f"https://answerous.contactous.com:9001/my-library?tenant_domain={tenant_domain}&user_id={user_id}&lms_platform_id={lms_platform_id}" return RedirectResponse(url=redirect_url, status_code=302) # return {"user": token.get("user"), "access_token": token["access_token"] , "expires_in": token.get("expires_in"), "accounts": accounts, "account_id": account_id} async def validate_token( user_id: int | None = None, email: str | None = None, db: Session = Depends(dbase.get_db) ): if user_id: user = db.query(User).filter(User.id == user_id).first() elif email: user = db.query(User).filter(User.email == email).first() if user: plateform = db.query(LMSPlatform).filter(LMSPlatform.id == user.platform_id).first() oauth = get_oauth_client(plateform) if plateform: access_token_entry = db.query(AccessToken).filter(AccessToken.user_id == user.id).order_by(AccessToken.id.desc()).first() logger.info("User: %s", user.id) logger.info("Platform: %s", plateform.id) logger.info("Access Token: %s", access_token_entry.id) logger.info("API token: %s", access_token_entry.rest_api_token if access_token_entry else "N/A") if access_token_entry: access_token = access_token_entry.rest_api_token if access_token and access_token_entry.expires_at < datetime.utcnow(): if not access_token_entry.refresh_token: return {"access_token": None, "error": "No access token", "redirect_url": f"{config('APP_URL')}/user/login?tenant_domain={plateform.tenant_domain}&user_id={user.id}&lms_platform_id={plateform.id}"} # Token expired new_token = await oauth.canvas_dynamic.fetch_access_token( grant_type="refresh_token", refresh_token=access_token_entry.refresh_token, ) logger.info("Refreshed token: %s", new_token) access_token = new_token["access_token"] access_token_entry.rest_api_token = new_token["access_token"] if "refresh_token" in new_token: access_token_entry.refresh_token = new_token["refresh_token"] access_token_entry.expires_at = datetime.utcnow() + timedelta(seconds=new_token.get("expires_in", 3600)) db.commit() return {"access_token": new_token["access_token"], "expires_in": new_token.get("expires_in")} elif access_token: return {"access_token": access_token, "expires_in": (access_token_entry.expires_at - datetime.utcnow()).total_seconds()} else: access_token = None else: access_token = None if not access_token: # return await login(Request, plateform.tenant_domain, user.id, plateform.id) return {"access_token": access_token, "error": "No access token", "redirect_url": f"{config('APP_URL')}/user/login?tenant_domain={plateform.tenant_domain}&user_id={user.id}&lms_platform_id={plateform.id}"} return None async def get_developer_keys(access_token: str, tenant_domain: str, account_id: int): headers = {"Authorization": f"Bearer {access_token}"} resp = httpx.get(f"{tenant_domain}/api/v1/accounts/{account_id}/developer_keys", headers=headers) if resp.status_code != 200: # raise HTTPException(status_code=resp.status_code, detail=resp.text) return errorResponse( resp.text, resp.status_code ) return resp.json() async def get_course_creation_accounts(access_token: str, tenant_domain: str): headers = {"Authorization": f"Bearer {access_token}"} resp = httpx.get(f"{tenant_domain}/api/v1/course_creation_accounts", headers=headers) if resp.status_code != 200: # raise HTTPException(status_code=resp.status_code, detail=resp.text) return errorResponse( resp.text, resp.status_code ) return resp.json() async def get_current_token( request: Request, credentials: HTTPAuthorizationCredentials = Depends(bearer) ): if credentials: return credentials.credentials session_token = request.session.get("token", {}).get("access_token") if session_token: return session_token return errorResponse( "Missing access token", 401 ) # raise HTTPException(status_code=401, detail="Missing access token") @router.get("/courses") async def list_courses(token: str = Depends(get_current_token)): headers = {"Authorization": f"Bearer {token}"} resp = httpx.get(f"{config('CANVAS_BASE_URL')}/api/v1/courses", headers=headers) if resp.status_code != 200: raise HTTPException(status_code=resp.status_code, detail=resp.text) return resp.json() @router.post("/create_course") async def create_course( account_id: int, course_name: str, token: str = Depends(get_current_token) ): # access_token = request.session.get("token", {}).get("access_token") if not token: # raise HTTPException(status_code=401, detail="Missing access token") return errorResponse( "Missing access token", 401 ) url = f"{config('CANVAS_BASE_URL')}/api/v1/accounts/{account_id}/courses" headers = {"Authorization": f"Bearer {token}"} data = {"course[name]": course_name, "course[is_public]": False} async with httpx.AsyncClient() as client: response = await client.post(url, headers=headers, data=data) if response.status_code != 200: # raise HTTPException(status_code=response.status_code, detail=response.text) return errorResponse( response.text, response.status_code ) return response.json() @router.post('/content_migration') async def content_migrations( course_id: int, imscc_path: str, token: str = Depends(get_current_token), db: Session = Depends(dbase.get_db) ): if not token: raise HTTPException(status_code=401, detail="Missing access token") dbcourse_id = 162 headers = {"Authorization": f"Bearer {token}"} init_url = f"{config('CANVAS_BASE_URL')}/api/v1/courses/{course_id}/content_migrations" course = db.query(Courses).filter(Courses.id == dbcourse_id).first() if not course: return errorResponse( "Course not found." ) topic = course.topic if not topic: return errorResponse( "Course topic is required to generate IMSCC." ) article_title = truncate_filename( topic.replace(" ", "_").replace("/", "_") ) filename = f"{article_title}.imscc" upload_info = await upload_file_to_canvas(course_id, imscc_path, token) print(upload_info) init_data = { "pre_attachment[name]": filename, "migration_type": "common_cartridge_importer", # IMSCC files are ZIPs "settings[file_url]": upload_info['upload_info']['upload_url'] } migration_info = {} async with httpx.AsyncClient() as client: init_resp = await client.post(init_url, headers=headers, data=init_data) if init_resp.status_code != 200: raise HTTPException(status_code=init_resp.status_code, detail=init_resp.text) migration_info = init_resp.json() print(migration_info) return {"status": "success", "upload_info": upload_info, "migration_info": migration_info} @router.post("/upload_files") async def upload_imscc( course_id: int, imscc_path: str, token: str = Depends(get_current_token) ): if not token: raise HTTPException(status_code=401, detail="Missing access token") upload_info = await upload_file_to_canvas(course_id, imscc_path, token) return {"status": "success", "upload_info": upload_info} async def upload_file_to_canvas(course_id: int, imscc_path: str, token: str): headers = {"Authorization": f"Bearer {token}"} # Step 1: Request upload slot init_url = f"{config('CANVAS_BASE_URL')}/api/v1/courses/{course_id}/files" init_data = { "name": os.path.basename(imscc_path), "content_type": "application/zip", # IMSCC files are ZIPs "on_duplicate": "rename", } async with httpx.AsyncClient() as client: init_resp = await client.post(init_url, headers=headers, data=init_data) if init_resp.status_code != 200: # raise HTTPException(status_code=init_resp.status_code, detail=init_resp.text) return errorResponse( init_resp.text ) upload_info = init_resp.json() # Step 2: Upload file to the returned URL upload_url = upload_info["upload_url"] upload_params = upload_info["upload_params"] async with aiofiles.open(imscc_path, "rb") as f: file_data = await f.read() # print("Uploading to:", upload_url) # print("With params:", upload_params) # print("File size:", len(file_data)) upload_resp = await client.post(upload_url, data=upload_params, files={"file": file_data}) if upload_resp.status_code not in (200, 201, 204, 303): # raise HTTPException(status_code=upload_resp.status_code, detail=upload_resp.text) return errorResponse( upload_resp.text, upload_resp.status_code ) return {"status": "success", "upload_info": upload_info} # @router.get("/refresh") # async def refresh_access_token(request: Request): # token = request.session.get("token") # print(token) # if not token or "refresh_token" not in token: # raise HTTPException(status_code=400, detail="No refresh token") # new_token = await oauth.canvas.fetch_access_token( # grant_type="refresh_token", # refresh_token=token["refresh_token"], # ) # request.session["token"] = new_token # return {"access_token": new_token["access_token"], "expires_in": new_token.get("expires_in")} async def get_self_profile(tenant_domain: str, token: str): headers = {"Authorization": f"Bearer {token}"} url = f"{tenant_domain}/api/v1/users/self" async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers) if response.status_code != 200: # raise HTTPException(status_code=response.status_code, detail=response.text) return errorResponse( response.text, response.status_code ) return response.json() @router.get('/find-account_id') async def get_account_id_from_courses(access_token: str, user_id: int): headers = {"Authorization": f"Bearer {access_token}"} print(user_id) url = f"{config('CANVAS_BASE_URL')}/api/v1/users/{user_id}/courses" async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers) response.raise_for_status() courses = response.json() if not courses: raise Exception("No courses found for this user.") # take the first course’s account_id return courses[0]["account_id"] @router.get("/migration_status") async def get_migration_status( course_id: int, migration_id: int, wait: bool = False, # optional: whether to poll until completion access_token: str = Depends(get_current_token), ): """ Check the status of a content migration in Canvas. If `wait=True`, polls the API every 3 seconds until the migration is finished. """ headers = {"Authorization": f"Bearer {access_token}"} status_url = f"{config('CANVAS_BASE_URL')}/api/v1/courses/{course_id}/content_migrations/{migration_id}" async with httpx.AsyncClient() as client: while True: response = await client.get(status_url, headers=headers) if response.status_code != 200: raise HTTPException(status_code=response.status_code, detail=response.text) migration = response.json() workflow_state = migration.get("workflow_state") progress_url = migration.get("progress_url") # Optional: Fetch more detailed progress info progress = None if progress_url: progress_resp = await client.get(progress_url, headers=headers) if progress_resp.status_code == 200: progress = progress_resp.json() result = { "migration_id": migration_id, "workflow_state": workflow_state, "progress": progress, "migration_info": migration, } # If not waiting, or the migration is finished, return if not wait or workflow_state in ("completed", "failed"): return result # Otherwise, poll every 3 seconds await asyncio.sleep(3) async def create_and_migrate_course( course_id: str, account_id: int, course_name: str, s3_imscc_url: str, # Public or signed AWS S3 URL token, plateform: LMSPlatform, db: Session = Depends(dbase.get_db) ): """ 1. Creates a Canvas course 2. Runs a content migration using IMSCC from AWS S3 URL (no file upload needed) """ if not token: # raise HTTPException(status_code=401, detail="Missing access token") return JSONResponse(status_code=401, content={"error": "Missing access token"}) headers = {"Authorization": f"Bearer {token}"} try: # ------------------------------------ # STEP 1 — Self profile (validate token) # ------------------------------------ self_profile_response = await get_self_profile(plateform.tenant_domain,token) # logger.info("Fetched user profile: %s", self_profile_response) canvas_user_id = self_profile_response["id"] # logger.info("Canvas User ID: %s", canvas_user_id) # ------------------------------------ # STEP 2 — Create the Canvas course # ------------------------------------ existing_course = db.query(Courses).filter(Courses.course_id == course_id).first() if existing_course.canvas_course_id: new_course_id = existing_course.canvas_course_id # logger.info("Course already exists with ID: %s", new_course_id) get_course_details_url = f"{plateform.tenant_domain}/api/v1/courses/{new_course_id}" async with httpx.AsyncClient() as client: course_response = await client.get(get_course_details_url, headers=headers) if course_response.status_code != 200: # raise HTTPException(status_code=course_response.status_code, detail=course_response.text) return JSONResponse(status_code=course_response.status_code, content={"error": course_response.text}) if course_response.status_code == 404: create_course_url = f"{plateform.tenant_domain}/api/v1/accounts/{account_id}/courses" course_data = { "course[name]": course_name, "course[is_public]": False, } async with httpx.AsyncClient() as client: create_response = await client.post(create_course_url, headers=headers, data=course_data) if create_response.status_code != 200: # raise HTTPException(status_code=create_response.status_code, detail=create_response.text) return JSONResponse(status_code=create_response.status_code, content={"error": create_response.text}) created_course = course_response.json() new_course_id = created_course["id"] # logger.info("Fetched existing course details: %s", created_course) else: create_course_url = f"{plateform.tenant_domain}/api/v1/accounts/{account_id}/courses" course_data = { "course[name]": course_name, "course[is_public]": False, } async with httpx.AsyncClient() as client: create_response = await client.post(create_course_url, headers=headers, data=course_data) if create_response.status_code != 200: # raise HTTPException(status_code=create_response.status_code, detail=create_response.text) return JSONResponse(status_code=create_response.status_code, content={"error": create_response.text}) created_course = create_response.json() new_course_id = created_course["id"] # logger.info("Created course: %s", created_course) # logger.info("Created course ID: %s", new_course_id) # ------------------------------------ # STEP 3 — Create content migration # using AWS S3 public URL (Recommended) # ------------------------------------ # logger.info("Starting content migration for course ID: %s using IMSCC URL: %s", new_course_id, s3_imscc_url) migration_url = f"{plateform.tenant_domain}/api/v1/courses/{new_course_id}/content_migrations" # Canvas supports direct URL import without uploading file migration_data = { "migration_type": "common_cartridge_importer", "settings[file_url]": s3_imscc_url, # DIRECT S3 URL } async with httpx.AsyncClient() as client: migration_response = await client.post(migration_url, headers=headers, data=migration_data) if migration_response.status_code not in (200, 201): # raise HTTPException(status_code=migration_response.status_code, detail=migration_response.text) return JSONResponse(status_code=migration_response.status_code, content={"error": migration_response.text}) migration_info = migration_response.json() # logger.info("Started content migration: %s", migration_info) course = db.query(Courses).filter(Courses.course_id == course_id).first() course.canvas_course_id = new_course_id db.commit() db.refresh(course) await enroll_teacher(new_course_id, canvas_user_id, plateform.tenant_domain, token) lms_course_log = LMSCourseLog( course_id=course.id, lms_course_id=new_course_id, course_info=json.dumps(jsonable_encoder(created_course)), migration_info=json.dumps(jsonable_encoder(migration_info)) ) db.add(lms_course_log) db.commit() return { "status": "success", "new_course_id": new_course_id, "course_created": created_course, "migration_started": migration_info, } except Exception as e: logger.error("Error in create_and_migrate_course: %s", str(e)) # raise HTTPException(status_code=500, detail=str(e)) return JSONResponse(status_code=500, content={"error": str(e)}) async def enroll_teacher(course_id: int, user_id: int, tenant_domain: str, token: str): url = f"{tenant_domain}/api/v1/courses/{course_id}/enrollments" headers = {"Authorization": f"Bearer {token}"} data = { "enrollment[user_id]": user_id, "enrollment[type]": "TeacherEnrollment", "enrollment[enrollment_state]": "active", } async with httpx.AsyncClient() as client: resp = await client.post(url, headers=headers, data=data) if resp.status_code != 200: # raise HTTPException(status_code=resp.status_code, detail=resp.text) return JSONResponse(status_code=resp.status_code, content={"error": resp.text}) return resp.json()