course_evalutation_requests.py

import asyncio
import getpass
import re
import time
from typing import Coroutine
from httpx import AsyncClient, Response, ConnectError
# from requests import Response, AsyncClient
from course_enrollment import authenticate_recursively, extract_form_data, get_course_id_dict, remove_amp_string, retry
from icecream import ic

class EvaluationFailed(Exception):
   def __init__(self, course: str, status_code: int):
      message: str = f"{course} evaluation failed with status code {status_code}"
      super().__init__(message)  # Pass the message to the base Exception class
      
HEADERS = {
   "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,"
            "image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
   "accept-language": "en-US,en;q=0.9",
   "accept-encoding": "gzip, deflate, br, zstd",
   "dnt": "1",
   "sec-ch-ua": '"Microsoft Edge";v="137", "Chromium";v="137", "Not/A)Brand";v="24"',
   "sec-ch-ua-mobile": "?0",
   "sec-ch-ua-platform": '"Windows"',
   "sec-fetch-dest": "document",
   "sec-fetch-mode": "navigate",
   "sec-fetch-site": "none",
   "sec-fetch-user": "?1",
   "upgrade-insecure-requests": "1",
   "user-agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/137.0.0.0 Safari/537.36 Edg/137.0.0.0"),
}      
      
      
@retry(exceptions=Exception, tries=3, delay=0, backoff=0)
async def course_evaluation_recursive(course: str, course_id: str, evaluation_id: str, session: AsyncClient) -> None:
   try: 
      await course_evaluation(course, course_id, evaluation_id, session)
   except EvaluationFailed:
      await course_evaluation_recursive(course, course_id, evaluation_id, session)

      
async def course_evaluation(course: str, course_id: str, evaluation_id: str, session: AsyncClient) -> None:
   query: dict[str, str] = {
      "id": evaluation_id,
      "courseid": course_id
   }

   # Wrap the synchronous request in asyncio.to_thread
   request_evaluation_form: Response = await session.get(
      "https://moodle.cu.edu.ng/mod/feedback/complete.php",
      params=query
   )

   evaluation_confirmation_match = re.search(r"</button>\s*(?=You)([^.]*.)\s*</", request_evaluation_form.text)
   if evaluation_confirmation_match:
      print(evaluation_confirmation_match.group(1))
      print(f"{course} evaluation verified complete")
      return

   form_data: dict[str, str] = await extract_form_data(request_evaluation_form.text, n_tags=8)

   evaluation_url = "https://moodle.cu.edu.ng/mod/feedback/complete.php"

   answer_list: list[str] = [
      "1", "4", "4", "1", "1", "4", "4", "2", "3", "1",
      "1", "4", "4", "4", "1", "4", "4", "1", "1", "4",
      "1", "4", "4", "1", "4", "4", "1", "4", "1", " ",
      "4"
   ]

   answer_to_evaluation: dict[str, str] = {'savevalues': 'Submit your answers',
                                           '_qf__mod_feedback_complete_form': '1',
                                           'startitempos': '',
                                          'gopage': '0',
                                          'lastitempos': '',
                                          'lastpage': ''
                                          }
   pattern = r'\bmultichoice_[0-9]+\b|(?<=")textfield_[0-9]+\b'
   question_list = re.findall(pattern, request_evaluation_form.text)

   for question, answer in zip(question_list, answer_list):
      answer_to_evaluation[question] = answer

   form: dict[str, str] = {**form_data, **answer_to_evaluation}
   if form.get("redirect"):
      form.pop("redirect")
   # ic(form)
   # Wrap the synchronous POST request in asyncio.to_thread
   request: Response = await session.post(
      evaluation_url,
      # params=query,
      data=form)

   if request.status_code != 200:
      raise EvaluationFailed(course, request.status_code)

   print(f"{course} evaluation complete {request.status_code}")
      # print(f"error {request.status_code}")
      # with open("test.txt", "w") as file:
      #    file.write(str(request.text))
      
async def general_evaluation(session: AsyncClient):
   html: Response = await session.get("https://moodle.cu.edu.ng/?redirect=0")
   html_text = await remove_amp_string(html.text)
   pattern = r'<a.*?href=".*?mod/feedback/.*?\?id=([0-9]+)"'
   general_evalutation_id_list = re.findall(pattern, html_text)
   task_list = []
   for ID in general_evalutation_id_list:
      task = course_evaluation(course="general evaluation", course_id="", evaluation_id=ID, session=session)
      task_list.append(task)
   await asyncio.gather(*task_list)
   return general_evalutation_id_list
   
async def run(session: AsyncClient) -> None:
   
   # username: str = input("Username: ")
   # password: str = getpass.getpass("Password: ")
   number: int = int(input("No of users"))
   username_password_dict = {}
   for _ in range(number):
      username: str = input("Username: ")
      password: str = input("Password: ")
      username_password_dict[username] = password
   task = []
   for username, password in username_password_dict.items():
      task.append(course_enroll_per_user(session, username, password))
   await asyncio.gather(*task)
   t: float = time.perf_counter()
   # await authenticate_recursively(session=session, username="", password="")

   
   
   t2: float = time.perf_counter()
   ic(f"{t2 - t = }")

async def course_enroll_per_user(session, username, password):
   await authenticate_recursively(session=session, username=username, password=password)
   # evaluation_id_list = await general_evaluation(session)

   # course_dict: dict[str, str] = await get_course_id_dict(session=session)
   # print(evaluation_id_list)
   await course_evaluation(course="general evaluation", course_id="", evaluation_id="51912", session=session) 
   # task_list = []
   # for course, course_id in course_dict.items():
   #   task = course_evaluation_recursive(course=course, course_id=course_id, evaluation_id=evaluation_id_list[0], session=session)
   #   task_list.append(task)

   # concurrency_limit = 15
   # # Create a semaphore to limit concurrency
   # semaphore = asyncio.Semaphore(concurrency_limit)

   # # Create a list of tasks with semaphore control
   # task_list1 = [run_task_with_semaphore(task, semaphore) for task in task_list]
   # await asyncio.gather(*task_list1)

async def run_task_with_semaphore(task: Coroutine, semaphore: asyncio.Semaphore):
   async with semaphore:
      return await task
   
async def main() -> None:
   # limits = Limits(max_connections=20, max_keepalive_connections=20)
   # async with AsyncClient(follow_redirects=True, limits=limits, timeout=100) as session:
   async with AsyncClient(
      http2=True,
      headers=HEADERS,
      timeout=30.0,
      trust_env=False,
      follow_redirects=True
   ) as session:

      # Authenticate once
      await course_enroll_per_user(session=session, username="2201898", password="victory2004")

   
if "__main__" == __name__ :
   asyncio.run(main())