IGNOU Telegram Bot v1.0

bot/helper/extractor.py

@@ -0,0 +1,55 @@
+
+class User:
+    def __init__(self, user):
+        if user is None:
+            user = {}
+
+        self._id = user.get("_id")
+        self.name = user.get("name")
+        self.course = user.get("action", {}).get("course")
+        self.myenrollment = user.get("myenrollment")
+        self.following = user.get("following", {})
+        self.enrollment = user.get("action", {}).get("enrollment")
+        self.last_used_on = user.get("action", {}).get("last_used_on")
+        self.is_admin = user.get("role_status", {}).get("is_admin", False)
+        self.user = user
+
+    def dict(self):
+        return self.user
+
+#
+class Student:
+    def __init__(self, student):
+        if student is None:
+            student = {}
+
+        self._id = student.get("_id")
+        self.name = student.get("name")
+        self.course = student.get("course")
+        self.myenrollment = student.get("myenrollment")
+        self.followers = student.get("followers", {})
+        self.enrollment = student.get("action", {}).get("enrollment")
+        self.last_used_on = student.get("action", {}).get("last_used_on")
+        self.is_admin = student.get("role_status", {}).get("is_admin", False)
+
+        self.grade = self.Grade(student.get("grade"))
+        self.tee = self.Tee(student.get("tee"))
+
+        self.student = student
+
+    def dict(self):
+        return self.student
+
+    # grade card
+    class Grade:
+        def __init__(self, grade):
+            self.passed = grade.get("count",{}).get("passed", 0)
+            self.failed = grade.get("count",{}).get("failed", 0)
+            self.checked = grade.get("checked")
+
+    # tee card
+    class Tee:
+        def __init__(self, tee):
+            self.count = tee.get("count",0)
+            self.checked = tee.get("checked")
+

bot/helper/ignoubooks.py

@@ -0,0 +1,77 @@
+import requests
+import os
+
+class IgnouBooks:
+
+    def __init__(self,course=None,subject='None') -> None:
+        self.headers = {'User-Agent':'Dalvik/2.1.0 (Linux; U; Android 10; Redmi Note 7 Pro Build/QQ3A.200605.001)','b8S3lCfLoGo4DVAzNQnl5OpMyALyq8e2WpWbZTMlwZ5iPHr.UaVNW':'J0lta3zy@19' }
+        self.course = course.upper()  # Course Name like BCA or MCA
+        self.subject = subject.upper()  # If subject code passed like BCS011 or MCS023
+        self.pId = '' # IGNOU course id like BCA 23 and MCA 25
+        self.cId = '' # IGNOU Course subject id list
+        self.courseList = '' # get all subject of course
+
+    def courseCode(self):
+        courses = ['https://egkapp.ignouonline.ac.in/api/programmes/type/Bachelor','https://egkapp.ignouonline.ac.in/api/programmes/type/Master'] 
+        for courseurl in courses:
+            response = requests.get(courseurl,headers = self.headers).json()
+            for res in response:
+                if '(' not in res['pCode']:
+                    if self.course == res['pCode'].upper():
+                        self.pId = res['pId']
+                        return
+                elif '(' in res['pCode'] and 'English' == res['pMedium']:
+                    if self.course == res['pCode'].split("(")[0].upper():
+                        self.pId = res['pId']
+                        return
+
+    def subjectCode(self):
+        self.courseCode()
+        subjects = 'https://egkapp.ignouonline.ac.in/api/courses/p/' + str(self.pId)
+
+        response = requests.get(subjects,headers = self.headers).json()
+
+        substring =''  # 〽️
+        
+        for res in response:
+            fullname = res['cCode']
+            con = any(map(str.isdigit, fullname))
+            if not con:
+                con = False
+                continue
+            if '-' in fullname:
+                first = fullname.split('-')[0]
+                last = str(int(fullname.split('-')[1])//1)
+                finalname = first + last
+            
+            if finalname == self.subject:
+                self.cId = res['cId']
+            title = res['cTitle']
+            coursename = res['cpId']['pCode']
+            substring += f'{title} \n〽️ `/book {coursename} {finalname}`\n\n'
+        
+        self.courseList =  substring
+
+    def getCourseSubjectlist(self):
+        self.courseCode()
+        self.subjectCode()
+        return self.courseList
+
+    def getDownload(self):
+        self.courseCode()
+        self.subjectCode()
+        downloads = 'https://egkapp.ignouonline.ac.in/api/blocks/c/' + str(self.cId)
+
+        response = requests.get(downloads,headers = self.headers).json()
+
+        namelist = []  
+        for res in response:
+            downurl = 'https://egkapp.ignouonline.ac.in/api/blocks/download/' + str(res['bId'])
+
+            downloadresponse = requests.get(downurl,headers = self.headers)
+            
+            open(f"{res['bCode']} [{self.subject}][@IGNOUpyBoT].pdf", 'wb').write(downloadresponse.content)
+            namelist.append(f"{res['bCode']} [{self.subject}][@IGNOUpyBoT].pdf")
+            print(f"{res['bCode']} [{self.subject}][@IGNOUpyBoT].pdf")
+
+        return namelist

bot/helper/ignoucrawler.py

@@ -0,0 +1,233 @@
+import asyncio
+from bot.helper.ignouresult import IgnouResult
+from bot.database import Database
+from bot.helper.extractor import Student
+import datetime
+import time
+
+from pyrogram import Client
+from pyrogram.errors import FloodWait,PeerIdInvalid
+
+db = Database()
+
+
+class IgnouCrawler:
+
+    def __init__(self,client) -> None:
+
+        self.db = db
+        self.client = client
+
+        self.greeted = {
+            "grade" : {},
+            "tee" : {}
+        }
+        self.greet_msg = {
+            "grade" : "Grade Card Updated Today ",
+            "tee" : "One more result out Today 🤒"
+        }
+
+        self.todayDate = datetime.datetime.today().strftime('%B %d, %Y')
+
+    async def greet_user(self, result_type, user_id):
+
+        if not self.greeted.get(result_type).get(user_id):
+            self.greeted[result_type][user_id] = True
+            try:
+                await self.client.send_message(
+                    user_id,
+                    f"<b>{self.greet_msg.get(result_type)}👩🏻‍🎨</b>",parse_mode='html')
+            except FloodWait as e:
+                time.sleep(e.x)
+                await self.client.send_message(
+                    user_id,
+                    f"<b>{self.greet_msg.get(result_type)}👩🏻‍🎨</b>",parse_mode='html')
+            except PeerIdInvalid as e:
+                print(f"{user_id} -> {e}")
+
+    async def teeCrawl(self, student: Student):
+
+        data = IgnouResult('roshan'+ student._id).teeResultString()
+
+        if data and student.tee.count != data.get("count"):
+
+            title = '<pre>' + f'Name : {student.name} -> {student.course}\n' + '</pre>'
+
+            for user_id in student.followers:
+
+                if not self.greeted.get("tee").get(user_id):
+                    await self.greet_user("tee", user_id)
+
+                try:
+                    await self.client.send_message(
+                        chat_id=user_id,
+                        text= title + data.get("result"),
+                        parse_mode='html')
+
+                except FloodWait as e:
+                    time.sleep(e.x)
+                    await self.client.send_message(
+                        chat_id=user_id,
+                        text= title + data.get("result"),
+                        parse_mode='html')
+                except PeerIdInvalid as e:
+                    print(f"{user_id} -> {e}")
+
+            await self.db.update(
+                self.db.crawler,
+                student._id,
+                {
+                    "$set": {
+                        "tee.count": data.get("count"),
+                        "tee.checked": self.todayDate
+                    }
+                }
+            )
+
+    async def teeTask(self):
+
+        print("Tee Result Crawling : {}".format(datetime.datetime.today().strftime("%d/%m/%Y %H:%M:%S")))
+
+        await self.db.update(
+            self.db.site,
+            "ignou",
+            {
+                "$set" : {
+                    "tee_checked": datetime.datetime.today().strftime("%d/%m/%Y %H:%M:%S")
+                }
+            }
+        )
+
+        tasks = []
+
+        students = await db.get_all_crawlers()
+        self.greeted['tee'] = {}
+
+        # Check first TEE SIte Updated or not
+        site = await IgnouResult().teeCardUpdated()
+        if not site.get("updated"):
+            # print("Tee card Site not Updated")
+            return
+
+        # if site updated check for results
+        async for student in students:
+            student_info = Student(student)
+
+            if student_info.tee.checked == self.todayDate:
+                continue
+
+            tasks.append(
+                    asyncio.create_task(
+                        self.teeCrawl(student_info)
+                    )
+                )
+
+        await asyncio.gather(*tasks)
+
+        await self.db.update(
+            self.db.site,
+            "ignou",
+            {
+                "$set" : {
+                    "tee" : site.get("date"),
+                    "tee_checked": datetime.datetime.today().strftime("%d/%m/%Y %H:%M:%S")
+                }
+            }
+        )
+
+    async def gradeCrawl(self,student: Student):
+
+        data = IgnouResult(student.course + student._id).gradeResultString()
+
+        grade_passed = data.get("json", {}).get("count", {}).get("passed", 0)
+        grade_failed = data.get("json", {}).get("count", {}).get("failed", 0)
+
+        if data and (int(student.grade.passed) !=  grade_passed or int(student.grade.failed) != grade_failed) or True:
+
+            for user_id in student.followers:
+
+                if not self.greeted.get("grade").get(user_id):
+                    await self.greet_user("grade", user_id)
+
+                try:
+                    await self.client.send_message(
+                        chat_id= user_id,
+                        text = data.get("result"),
+                        parse_mode ='html')
+
+                except FloodWait as e:
+                    time.sleep(e.x)
+                    await self.client.send_message(
+                        chat_id= user_id,
+                        text = data.get("result"),
+                        parse_mode ='html')
+                except PeerIdInvalid as e:
+                    print(f"{user_id} -> {e}")
+
+            await self.db.update(
+                self.db.crawler,
+                student._id,
+                {
+                    "$set" : {
+                        "grade.count.passed": grade_passed,
+                        "grade.count.failed": grade_failed,
+                        "grade.checked": self.todayDate
+                        }
+                }
+            )
+
+    async def gradeTask(self):
+        print("Grade Card Crawling : {}".format(datetime.datetime.today().strftime("%d/%m/%Y %H:%M:%S")))
+
+        await self.db.update(
+            self.db.site,
+            "ignou",
+            {
+                "$set": {
+                    "grade_checked": datetime.datetime.today().strftime("%d/%m/%Y %H:%M:%S")
+                }
+            }
+        )
+
+        tasks = []
+
+        students = await db.get_all_crawlers()
+        self.greeted['grade'] = {}
+
+        # Check first Grade Site Updated or not
+        site = await IgnouResult().gradeCardUpdated()
+        if not site.get("updated"):
+            # print("Grade card Site not Updated")
+            return
+
+        # if site updated check for results
+
+        async for student in students:
+            student_info = Student(student)
+
+            if student_info.grade.checked == self.todayDate:
+                continue
+
+            tasks.append(
+                    asyncio.create_task(
+                        self.gradeCrawl(student_info)
+                    )
+                )
+
+        await asyncio.gather(*tasks)
+
+        await self.db.update(
+            self.db.site,
+            "ignou",
+            {
+                "$set" : {
+                    "grade": site.get("date"),
+                    "grade_checked": datetime.datetime.today().strftime("%d/%m/%Y %H:%M:%S")
+                }
+            }
+        )
+
+
+if __name__ == "__main__":
+    pass
+

bot/helper/ignouresult.py

@@ -0,0 +1,218 @@
+import requests
+from bs4 import BeautifulSoup
+from prettytable import PrettyTable
+import re
+from bot.database import  Database
+
+db = Database()
+
+class IgnouResult:
+
+    def __init__(self,text='Bca 197940316') -> None:
+        self.text = text.upper().strip()
+        self.enrollmentNo = ''
+        self.courseId = ''
+        self.extractCourseIDandEnrollmentNo()
+        self.sem = 'Dec20'
+
+        self.session = requests.Session()
+        self.session.headers.update({
+            'Connection': 'keep-alive',
+            'Cache-Control': 'max-age=0',
+            'Upgrade-Insecure-Requests': '1',
+            'Content-Type': 'application/x-www-form-urlencoded',
+            'User-Agent': 'Mozilla/5.0 (Linux; Android 6.0.1; Moto G (4)) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.96 Mobile Safari/537.36',
+            '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.9',
+            'Sec-GPC': '1',
+            'Sec-Fetch-Site': 'same-origin',
+            'Sec-Fetch-Mode': 'navigate',
+            'Sec-Fetch-User': '?1',
+            'Sec-Fetch-Dest': 'document',
+            'Accept-Language': 'en-US,en;q=0.9,ru;q=0.8',
+        }
+
+        )
+    def extractCourseIDandEnrollmentNo(self):
+
+        self.courseId = re.findall("\D+", self.text)[0].strip()
+        self.enrollmentNo = re.findall('\d+', self.text)[0].strip()
+        return self.courseId,self.enrollmentNo
+
+    def teeResultJson(self):
+
+        data = {
+        'eno': self.enrollmentNo,
+        'myhide': 'OK'
+        }
+        teeUrl = f'https://termendresult.ignou.ac.in/TermEnd{self.sem}/TermEnd{self.sem}.asp'
+        response = self.session.post(teeUrl, data=data)
+        
+        if 'Not found' in response.text:
+            return {'status':'notok'} # result not found
+
+        soup = BeautifulSoup(response.text,'lxml')
+
+        trs = soup.find_all('tr')[1:]
+        resultJson = list()
+
+        for tr in trs:
+            info = tr.find_all("td")
+            data = list()
+            for index,col in enumerate(info):
+                if index == len(info) - 1:
+                    break
+                data.append(info[index].text.strip())
+            resultJson.append(data)
+        
+        return {'result' : resultJson,'count' : len(trs),'html':response.text, 'status':'ok'}
+
+
+
+
+    def gradeResultJson(self):
+
+        data = {
+        'Program': self.courseId,
+        'eno': self.enrollmentNo,
+        'submit': 'Submit',
+        'hidden_submit': 'OK'
+        }
+
+        if self.courseId in ['BCA', 'MCA', 'MP', 'MBP', 'PGDHRM', 'PGDFM', 'PGDOM', 'PGDMM', 'PGDFMP']:
+            url = 'https://gradecard.ignou.ac.in/gradecardM/Result.asp'
+        elif self.courseId in ['ASSSO', 'BA', 'BCOM', 'BDP', 'BSC']:
+            url = 'https://gradecard.ignou.ac.in/gradecardB/Result.asp'
+        else:
+            url = 'https://gradecard.ignou.ac.in/gradecardR/Result.asp'
+
+        response = self.session.post(url, data=data)
+
+        soup = BeautifulSoup(response.text,'lxml')
+
+        if 'Not found' in response.text or response.status_code != 200:
+            return {'status':'notok'} # result not found
+
+        trs = soup.find_all("tr")[1:]
+        
+        resultJson = list()
+        student = dict()
+        btags = soup.find_all('b')[3:6]
+
+        for n,ba in enumerate(btags,1):
+            if n == 1:
+                student['enrollment'] = ba.text.split(":")[1].strip()
+            elif n == 2:
+                student['name'] = ba.text.split(":")[1].strip()
+            else:
+                student['course'] = ba.text.split(":")[1].strip()
+
+        count = dict()
+        count['passed'] = 0
+        count['failed'] = 0
+        count['total'] = 0
+        for tr in trs:
+            data = list()
+
+            td = tr.find_all("td")
+
+            data.append(td[0].string.strip())
+            data.append(td[1].string.strip())
+            data.append(td[2].string.strip())
+            data.append(td[6].string.strip())
+
+            status = True if 'Not' not in td[7].string.strip() else False # ✅ ❌
+            data.append(status)
+
+            if status:
+                count['passed'] += 1
+            else:
+                count['failed'] += 1
+            count['total'] += 1
+            
+            resultJson.append(data)
+        
+        return {'result' :  resultJson,'student' : student,'count' : count,'html':response.text,'status':'ok'}
+    
+    def gradeResultString(self):
+        x = PrettyTable()
+        x.field_names = ["Course","Asign","Lab","Term","Status"]
+        
+        gradeJson = self.gradeResultJson()
+
+        if gradeJson['status'] != 'ok':
+            return False
+        header = 'Name : {}\n -> {} -> {}\n'.format(gradeJson['student']['name'],self.courseId,self.enrollmentNo)
+
+        for sub in gradeJson['result']:
+            tick = '✅' if sub[-1] else '❌'
+            sub[-1] =  tick
+            x.add_row(sub)
+
+        footer = ['count','T:{}'.format(gradeJson['count']['total']),
+                'P:{}'.format(gradeJson['count']['passed']),
+                'F:{}'.format(gradeJson['count']['failed']),
+                'L:{}'.format(gradeJson['count']['total']-gradeJson['count']['passed'])]
+        x.add_row(footer)
+        
+        return {
+            'enrollmentno' : self.enrollmentNo,
+            'course' : self.courseId,
+            'result' : '<pre>' + header + x.get_string() + '</pre>',
+            'json' : gradeJson
+        }
+
+    def teeResultString(self):
+        x = PrettyTable()
+        header = 'Enrollment no : {} ({})\n'.format(self.enrollmentNo,self.sem)
+        x.field_names = ["Course","Marks","Max","Month","Updation"]
+
+        teeJson = self.teeResultJson()
+
+        if teeJson['status'] != 'ok':
+            return False
+
+        for sub in teeJson['result']:
+            x.add_row(sub)
+
+        return {
+            'enrollmentno' : self.enrollmentNo,
+            'course' : self.courseId,
+            'count' : teeJson['count'],
+            'result' : '<pre>' + header + x.get_string() + '</pre>',
+            'json' : teeJson
+        }
+
+    async def gradeCardUpdated(self):
+        
+        response = self.gradeResultJson()
+
+        if response['status'] != 'ok':
+            return False
+
+        updated = re.findall('([\w]+ ?[\d]+, ?[\d]+)',response['html'])[0]
+
+        last_updateed = await db.get_site_update("ignou")
+
+        if updated != last_updateed.get("grade",''):
+            return {"date" : updated,"updated" : True}
+        return {"date" : updated,"updated" : False}
+
+
+    async def teeCardUpdated(self):
+
+        teeUrl = f'https://termendresult.ignou.ac.in/TermEnd{self.sem}/TermEnd{self.sem}.asp'
+
+        response = self.session.post(teeUrl)
+
+        if response.status_code != 200:
+            return False
+
+        updated = re.findall('([\w]+ ?[\d]+, ?[\d]+)',response.text)[0]
+
+        # todayDate = datetime.datetime.today().strftime('%B %d, %Y')
+
+        last_updateed = await db.get_site_update("ignou")
+
+        if updated != last_updateed.get("tee",''):
+            return {"date" : updated,"updated" : True}
+        return {"date" : updated,"updated" : False}