added flask app

.gitignore

@@ -7,4 +7,5 @@ data.txt
 *.ics
 .idea/
 .vscode
-venv
+venv
+.env

app.py

@@ -0,0 +1,212 @@
+"""
+This file contains the Flask application that serves as the backend for GYFT.
+"""
+import io
+import logging
+from typing import Dict, List
+import requests
+from flask import Flask, request, send_file, jsonify
+from iitkgp_erp_login import erp
+import iitkgp_erp_login.utils as erp_utils
+from flask_cors import CORS
+from timetable import generate_ics
+from gyft import get_courses
+
+
+app = Flask(__name__)
+CORS(app)
+
+headers = {
+    "timeout": "20",
+    "User-Agent":
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3",
+}
+
+
+def check_missing_fields(all_fields: Dict[str, str]) -> List[str]:
+    return [field for field, value in all_fields.items() if not value]
+
+
+class ErpResponse:
+    def __init__(
+        self,
+        success: bool = True,
+        message: str = None,
+        data: dict = None,
+        status_code: int = 200,
+    ):
+        self.success = success
+        self.message = message
+        self.data = data or {}
+        self.status_code = status_code
+
+        if not success:
+            logging.error(" %s", message)
+
+    def to_dict(self):
+        response = {"status": "success" if self.success else "error"}
+        if self.message:
+            response["message"] = self.message
+        if self.data:
+            response |= self.data
+        return response
+
+    def to_response(self):
+        return jsonify(self.to_dict()), self.status_code
+
+
+@app.route("/secret-question", methods=["POST"])
+def get_secret_question():
+    try:
+        data = request.form
+        all_fields = {
+            "roll_number": data.get("roll_number"),
+        }
+        missing = check_missing_fields(all_fields)
+        if len(missing) > 0:
+            return ErpResponse(
+                False, f"Missing Fields: {', '.join(missing)}", status_code=400
+            ).to_response()
+
+        session = requests.Session()
+        secret_question = erp.get_secret_question(
+            headers=headers,
+            session=session,
+            roll_number=all_fields["roll_number"],
+            log=True,
+        )
+        sessionToken = erp_utils.get_cookie(session, "JSESSIONID")
+
+        return ErpResponse(
+            True,
+            data={"SECRET_QUESTION": secret_question,
+                  "SESSION_TOKEN": sessionToken},
+        ).to_response()
+    except erp.ErpLoginError as e:
+        return ErpResponse(False, str(e), status_code=401).to_response()
+    except Exception as e:
+        return ErpResponse(False, str(e), status_code=500).to_response()
+
+
+@app.route("/request-otp", methods=["POST"])
+def request_otp():
+    try:
+        data = request.form
+        all_fields = {
+            "roll_number": data.get("roll_number"),
+            "password": data.get("password"),
+            "secret_answer": data.get("secret_answer"),
+            "sessionToken": request.headers["Session-Token"],
+        }
+        missing = check_missing_fields(all_fields)
+        if len(missing) > 0:
+            return ErpResponse(
+                False, f"Missing Fields: {', '.join(missing)}", status_code=400
+            ).to_response()
+
+        login_details = erp.get_login_details(
+            ROLL_NUMBER=all_fields["roll_number"],
+            PASSWORD=all_fields["password"],
+            secret_answer=all_fields["secret_answer"],
+            sessionToken=all_fields["sessionToken"],
+        )
+
+        session = requests.Session()
+        erp_utils.set_cookie(session, "JSESSIONID", all_fields["sessionToken"])
+        erp.request_otp(
+            headers=headers, session=session, login_details=login_details, log=True
+        )
+
+        return ErpResponse(
+            True, message="OTP has been sent to your connected email accounts"
+        ).to_response()
+    except erp.ErpLoginError as e:
+        return ErpResponse(False, str(e), status_code=401).to_response()
+    except Exception as e:
+        return ErpResponse(False, str(e), status_code=500).to_response()
+
+
+@app.route("/login", methods=["POST"])
+def login():
+    try:
+        data = request.form
+        all_fields = {
+            "roll_number": data.get("roll_number"),
+            "password": data.get("password"),
+            "secret_answer": data.get("secret_answer"),
+            "otp": data.get("otp"),
+            "sessionToken": request.headers["Session-Token"],
+        }
+        missing = check_missing_fields(all_fields)
+        if len(missing) > 0:
+            return ErpResponse(
+                False, f"Missing Fields: {', '.join(missing)}", status_code=400
+            ).to_response()
+
+        login_details = erp.get_login_details(
+            ROLL_NUMBER=all_fields["roll_number"],
+            PASSWORD=all_fields["password"],
+            secret_answer=all_fields["secret_answer"],
+            sessionToken=all_fields["sessionToken"],
+        )
+        login_details["email_otp"] = all_fields["otp"]
+
+        session = requests.Session()
+        erp_utils.set_cookie(session, "JSESSIONID", all_fields["sessionToken"])
+        sso_token = erp.signin(
+            headers=headers, session=session, login_details=login_details, log=True
+        )
+
+        return ErpResponse(True, data={"ssoToken": sso_token}).to_response()
+    except erp.ErpLoginError as e:
+        return ErpResponse(False, str(e), status_code=401).to_response()
+    except Exception as e:
+        return ErpResponse(False, str(e), status_code=500).to_response()
+
+
+@app.route("/timetable", methods=["POST"])
+def download_ics():
+    try:
+        data = request.form
+        all_fields = {
+            "roll_number": data.get("roll_number"),
+            "ssoToken": request.headers["SSO-Token"],
+        }
+        missing = check_missing_fields(all_fields)
+        if len(missing) > 0:
+            return ErpResponse(
+                False, f"Missing Fields: {', '.join(missing)}", status_code=400
+            ).to_response()
+
+        roll_number = all_fields["roll_number"]
+        sso_token = all_fields["ssoToken"]
+
+        session = requests.Session()
+        erp_utils.populate_session_with_login_tokens(session, sso_token)
+
+        courses = get_courses(session, sso_token, roll_number)
+
+        print("Timetable fetched.\n")
+
+        ics_content = generate_ics(courses, "")
+
+        # Create an in-memory file-like object for the ics content
+        ics_file = io.BytesIO()
+        ics_file.write(ics_content.encode("utf-8"))
+        ics_file.seek(0)
+
+        return send_file(
+            ics_file,
+            as_attachment=True,
+            mimetype="text/calendar",
+            download_name=f"${roll_number}-timetable.ics",
+        )
+    except Exception as e:
+        return jsonify({"status": "error", "message": str(e)}), 500
+
+
+if __name__ == "__main__":
+    # Run the application on the local development server
+    app.run()
+
+# flask --app app.py run

gyft.py

@@ -1,14 +1,28 @@
 import argparse
+import iitkgp_erp_login.erp as erp
+import requests
 from timetable import delete_calendar, create_calendar, build_courses, generate_ics
-from utils import ERPSession
+from utils.dates import SEM_BEGIN
+
+
+headers = {
+    "timeout": "20",
+    "User-Agent":
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3",
+}
 
 
 def parse_args():
     parser = argparse.ArgumentParser()
-    parser.add_argument("-o", "--output",
-                        help="Output file containing timetable in .ics format")
-    parser.add_argument("-d", "--del-events", action="store_true",
-                        help="Delete events automatically added by the script before adding new events")
+    parser.add_argument(
+        "-o", "--output", help="Output file containing timetable in .ics format"
+    )
+    parser.add_argument(
+        "-d",
+        "--del-events",
+        action="store_true",
+        help="Delete events automatically added by the script before adding new events",
+    )
     args = parser.parse_args()
     return args
 
@@ -17,17 +31,22 @@ def main():
     args = parse_args()
     if args.del_events:
         delete_calendar()
-        if input("\nWould you like to generate a new timetable? (y/n): ").lower() == 'n':
+        if (
+            input("\nWould you like to generate a new timetable? (y/n): ").lower()
+            == "n"
+        ):
             return
 
     output_filename = args.output if args.output else "timetable.ics"
-    erp_session = ERPSession.login()
-    timetable_page = erp_session.post(erp_session.ERP_TIMETABLE_URL, cookies=True,
-                                      data=erp_session.get_timetable_details())
-    course_names = erp_session.get_course_names()
-    courses = build_courses(timetable_page.text, course_names)
 
-    print(f"Timetable fetched.\n")
+    session = requests.Session()
+    _, sso_token = erp.login(headers, session)
+
+    roll_number = erp.ROLL_NUMBER
+
+    courses = get_courses(session, sso_token, roll_number)
+
+    print("Timetable fetched.\n")
 
     print("What would you like to do now?")
     print("1. Add timetable directly to Google Calendar (requires client_secret.json)")
@@ -43,5 +62,50 @@ def main():
         exit()
 
 
+def get_courses(session: requests.Session, sso_token: str, roll_number: str):
+
+    erp_timetable_url = "https://erp.iitkgp.ac.in/Acad/student/view_stud_time_table.jsp"
+    courses_url: str = (
+        "https://erp.iitkgp.ac.in/Academic/student_performance_details_ug.htm?semno={}&rollno={}"
+    )
+
+    timetable_page = session.post(
+        headers=headers,
+        url=erp_timetable_url,
+        data={
+            "ssoToken": sso_token,
+            "module_id": "16",
+            "menu_id": "40",
+        },
+    )
+    sem_num = 1
+
+    if SEM_BEGIN.month > 6:
+        # autumn semester
+        sem_num = (int(SEM_BEGIN.strftime("%y")) -
+                   int(roll_number[:2])) * 2 + 1
+    else:
+        # spring semester - sem begin year is 1 more than autumn sem
+        sem_num = (int(SEM_BEGIN.strftime("%y")) - int(roll_number[:2])) * 2
+
+    r = session.post(
+        headers=headers,
+        url=courses_url.format(sem_num, roll_number),
+        data={
+            "ssoToken": sso_token,
+            "semno": sem_num,
+            "rollno": roll_number,
+            "order": "asc",
+        },
+    )
+    sub_dict = {item["subno"]: item["subname"] for item in r.json()}
+    course_names = {k: v.replace("&", "&") for k, v in sub_dict.items()}
+
+    print(course_names)
+
+    courses = build_courses(timetable_page.text, course_names)
+    return courses
+
+
 if __name__ == "__main__":
     main()

requirements.txt

@@ -2,7 +2,8 @@ beautifulsoup4==4.12.2
 google_api_python_client==2.90.0
 httplib2==0.22.0
 icalendar==5.0.7
-iitkgp_erp_login
+iitkgp_erp_login==2.4.2
 oauth2client==4.1.3
 pytz==2023.3
 Requests==2.31.0
+flask==3.0.3

timetable/generate_ics.py

@@ -51,7 +51,11 @@ def generate_ics(courses: list[Course], output_filename):
         event.add("dtstart", holiday[1])
         event.add("dtend", holiday[1] + timedelta(days=1))
         cal.add_component(event)
-
-    with open(output_filename, "wb") as f:
-        f.write(cal.to_ical())
-        print("\nYour timetable has been written to %s" % output_filename)
+
+
+    if output_filename != "":
+        with open(output_filename, "wb") as f:
+            f.write(cal.to_ical())
+            print("\nYour timetable has been written to %s" % output_filename)
+
+    return cal.to_ical().decode('utf-8')

utils/__init__.py

@@ -1,3 +1,3 @@
-from utils.network import *
 from utils.dates import *
 from utils.build_event import *
+

utils/dates.py

@@ -6,6 +6,7 @@
 import sys
 from collections import defaultdict
 
+
 SEM_BEGIN = build_event.generate_india_time(2024, 7, 22, 0, 0)
 MID_TERM_BEGIN = build_event.generate_india_time(2024, 9, 17, 0, 0)
 MID_TERM_END = build_event.generate_india_time(2024, 9, 25, 0, 0)