add webapp
Diff
web/db.json | 1 +
web/run.py | 6 ++++++
web/app/__init__.py | 24 ++++++++++++++++++++++++
web/app/auth.py | 21 +++++++++++++++++++++
web/app/dump.rdb | 0
web/app/job_manager.py | 16 ++++++++++++++++
web/app/main.py | 61 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
web/app/models.py | 36 ++++++++++++++++++++++++++++++++++++
web/app/jobs/scrape_cases.py | 68 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
web/app/modules/encryption.py | 33 +++++++++++++++++++++++++++++++++
web/app/modules/interface.py | 116 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
web/app/templates/base.html | 26 ++++++++++++++++++++++++++
web/app/templates/home.html | 34 ++++++++++++++++++++++++++++++++++
web/app/templates/login.html | 16 ++++++++++++++++
14 files changed, 458 insertions(+)
@@ -1,0 +1,1 @@
{"users": {"1": {"username": "admin", "password": "dontguessthisplzahh", "admin": true}}}
@@ -1,0 +1,6 @@
from app import create_app
app = create_app()
if __name__ == '__main__':
app.run(debug=True)
@@ -1,0 +1,24 @@
from flask import Flask
from flask_login import LoginManager
from .models import User
login_manager = LoginManager()
def create_app():
app = Flask(__name__)
app.secret_key = 'your_secret_key'
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
from .auth import auth as auth_blueprint
from .main import main as main_blueprint
app.register_blueprint(auth_blueprint)
app.register_blueprint(main_blueprint)
return app
@login_manager.user_loader
def load_user(user_id):
return User.get(user_id)
@@ -1,0 +1,21 @@
from flask import Blueprint, render_template, request, redirect, url_for
from flask_login import login_user
from .models import User
auth = Blueprint('auth', __name__)
@auth.route('/login', methods=['GET', 'POST'])
def login():
error = None
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
user = User.validate_login(username, password)
if user:
login_user(user)
return redirect(url_for('main.home'))
error = "Invalid credentials"
return render_template('login.html', error=error)
Binary files /dev/null and a/web/app/dump.rdb differ
@@ -1,0 +1,16 @@
from rq import Queue
from redis import Redis
from jobs.scrape_cases import scrape_cases
class JobManager:
def __init__(self):
redis = Redis()
self.q = Queue(connection=redis)
def enqueue_scrape(self, act, section, state_code):
return self.q.enqueue(
scrape_cases,
act,
section,
state_code
)
@@ -1,0 +1,61 @@
from flask import request, flash
from flask import Blueprint, render_template, redirect, url_for
from flask_login import login_required, logout_user, current_user
from .models import User
from .modules.interface import Interface
states = Interface().get_states()
main = Blueprint('main', __name__)
@main.route('/')
@login_required
def home():
return render_template('home.html', user=current_user, states=states)
@main.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('auth.login'))
@main.route('/create_user', methods=['POST'])
@login_required
def create_user():
username = request.form.get('username')
password = request.form.get('password')
if current_user.admin != True:
flash('Only admin can create new users.', 'error')
return redirect(url_for('main.home'))
if not username or not password:
flash('Username and password required.', 'error')
return redirect(url_for('main.home'))
user = User.create(username, password)
if user:
flash(f'User {username} created successfully.', 'success')
else:
flash(f'User {username} already exists.', 'error')
return redirect(url_for('main.home'))
@main.route('/enqueue_job', methods=['POST'])
@login_required
def enqueue_job():
act = request.form.get('act')
section = request.form.get('section')
state_code = request.form.get('state_code')
if not act or not state_code:
flash('All fields must be filled.', 'error')
return redirect(url_for('main.home'))
if not section:
section = ''
flash('Job created.', 'info')
return redirect(url_for('main.home'))
@@ -1,0 +1,36 @@
from flask_login import UserMixin
from tinydb import TinyDB, Query
db = TinyDB('db.json')
users_table = db.table('users')
UserQuery = Query()
class User(UserMixin):
def __init__(self, username, admin):
self.id = username
self.admin = admin
@staticmethod
def get(username):
result = users_table.get(UserQuery.username == username)
if result:
return User(username, result['admin'])
return None
@staticmethod
def validate_login(username, password):
user = users_table.get((UserQuery.username == username) & (UserQuery.password == password))
if user:
return User(username, user['admin'])
return None
@staticmethod
def create(username, password, admin=False):
if users_table.get(UserQuery.username == username):
return None
users_table.insert({'username': username, 'password': password, 'admin': admin})
return User(username, admin)
User.create('admin', 'dontguessthisplzahh', admin=True)
@@ -1,0 +1,68 @@
from modules.interface import Interface
from tinydb import TinyDB
import time
def scrape_cases(act, section, state_code, name=time.time_ns()):
db = TinyDB(f'{name}.json')
interface = Interface()
def get_act_number(acts):
for act_code, act_name in acts:
if act_name == act:
return act_code
return None
try:
districts = interface.get_districts(state_code)
except Exception as e:
print(f"[ERROR] Failed to scrape districts: {e}")
districts = []
for dist_code, dist_name in districts:
print(f'DISTRICT: {dist_name}')
try:
complexes = interface.get_complexes(state_code, dist_code)
except Exception as e:
print(f"[ERROR] Failed to scrape complexes for {dist_name}: {e}")
continue
for complex_code, complex_name in complexes:
print(f'COMPLEX: {complex_name}')
court_establishments = str(complex_code).split(',')
for i, court_establishment in enumerate(court_establishments, 1):
print(f'ESTABLISHMENT: {i}/{len(court_establishments)}')
try:
acts = interface.get_acts(state_code, dist_code, court_establishment)
act_number = get_act_number(acts)
except Exception as e:
print(f"[ERROR] Failed to scrape acts for complex {complex_name}: {e}")
continue
if not act_number:
continue
try:
cases = interface.search_by_act(state_code, dist_code, court_establishment, act_number, section)
except Exception as e:
print(f"[ERROR] Failed to scrape cases in complex {complex_name}: {e}")
continue
for j, case in enumerate(cases, 1):
print(f'CASE: {j}/{len(cases)}')
try:
case_no = case['case_no']
case_history = interface.case_history(state_code, dist_code, court_establishment, case_no)
except Exception as e:
print(f"[ERROR] Failed to get history for case {case.get('case_no', 'UNKNOWN')}: {e}")
continue
try:
case_history['case_no'] = case_no
case_history['complex_name'] = complex_name
db.insert(case_history)
except Exception as e:
print(f"[ERROR] Failed to parse orders for case {case_no}: {e}")
@@ -1,0 +1,33 @@
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
import base64
import os
import json
REQUEST_KEY = bytes.fromhex('4D6251655468576D5A7134743677397A')
RESPONSE_KEY = bytes.fromhex('3273357638782F413F4428472B4B6250')
GLOBAL_IV = "556A586E32723575"
IV_INDEX = '0'
RANDOMIV = os.urandom(8).hex()
IV = bytes.fromhex(GLOBAL_IV + RANDOMIV)
class Encryption:
@staticmethod
def encrypt(data):
cipher = AES.new(REQUEST_KEY, AES.MODE_CBC, IV)
padded_data = pad(json.dumps(data).encode(), 16)
ct = cipher.encrypt(padded_data)
ct_b64 = base64.b64encode(ct).decode()
return RANDOMIV + str(IV_INDEX) + ct_b64
@staticmethod
def decrypt(data):
data = data.strip()
iv_hex = data[:32]
ct_b64 = data[32:]
iv = bytes.fromhex(iv_hex)
ct = base64.b64decode(ct_b64)
cipher = AES.new(RESPONSE_KEY, AES.MODE_CBC, iv)
pt = unpad(cipher.decrypt(ct), 16)
return json.loads(pt.decode(errors="ignore"))
@@ -1,0 +1,116 @@
import requests
import os
from .encryption import Encryption
BASE_URL = "https://app.ecourts.gov.in/ecourt_mobile_DC"
RETRY_ATTEMPTS = 10
TIMEOUT = 5
class Interface:
def __init__(self):
self.token = self.fetch_token()
def fetch_token(self):
uid = os.urandom(8).hex() + ':in.gov.ecourts.eCourtsServices'
payload = Encryption.encrypt({"version": "3.0", "uid": uid})
r1 = requests.get(f"{BASE_URL}/appReleaseWebService.php", params={'params': payload})
token = Encryption.decrypt(r1.text)['token']
token = Encryption.encrypt(token)
if not token:
raise Exception
return token
def get(self, endpoint, data):
for _ in range(RETRY_ATTEMPTS):
try:
resp = requests.get(
f"{BASE_URL}/{endpoint}",
params={'params': data},
headers={"Authorization": f"Bearer {self.token}"},
timeout=TIMEOUT
)
return Encryption.decrypt(resp.text)
except:
continue
raise Exception
def get_states(self):
try:
data = Encryption.encrypt({'action_code': 'fillState'})
states_list = self.get('stateWebService.php', data)['states']
return list(map(lambda x: (x['state_code'], x['state_name']), states_list))
except RuntimeError:
raise Exception("Failed to scrape states")
def get_districts(self, state_code):
try:
data = Encryption.encrypt({"state_code": str(state_code)})
districts_list = self.get('districtWebService.php', data)['districts']
return list(map(lambda x: (x['dist_code'], x['dist_name']), districts_list))
except RuntimeError:
raise Exception("Failed to scrape districts")
def get_complexes(self, state_code, dist_code):
try:
data = Encryption.encrypt({
"action_code": "fillCourtComplex",
"state_code": str(state_code),
"dist_code": str(dist_code)
})
complexes_list = self.get('courtEstWebService.php', data)['courtComplex']
if complexes_list is None:
return []
return list(map(lambda x: (x['njdg_est_code'], x['court_complex_name']), complexes_list))
except RuntimeError:
raise Exception("Failed to scrape court complexes")
def get_acts(self, state_code, dist_code, complex_code):
try:
data = Encryption.encrypt({
"state_code": str(state_code),
"dist_code": str(dist_code),
"court_code": str(complex_code),
"searchText": "",
"language_flag": "english",
"bilingual_flag": "0"
})
acts_list = self.get('actWebService.php', data)['actsList'][0]['acts'].split('#')
return list(map(lambda x: (x.split('~')[0], x.split('~')[1]) if '~' in x else (x, None), acts_list))
except RuntimeError:
raise Exception("Failed to scrape acts")
def search_by_act(self, state_code, dist_code, complex_code, act_number, section=""):
try:
data = Encryption.encrypt({
"state_code": str(state_code),
"dist_code": str(dist_code),
"court_code_arr": str(complex_code),
"language_flag": "english",
"bilingual_flag": "0",
"selectActTypeText": str(act_number),
"underSectionText": section,
"pendingDisposed": "Disposed"
})
cases_list = self.get('searchByActWebService.php', data)
return cases_list['0']['caseNos']
except RuntimeError:
raise Exception("Failed to scrape cases by act")
def case_history(self, state_code, dist_code, complex_code, case_no):
try:
data = Encryption.encrypt({
"state_code": str(state_code),
"dist_code": str(dist_code),
"court_code": str(complex_code),
"language_flag": "english",
"bilingual_flag": "0",
"case_no": case_no
})
return self.get('caseHistoryWebService.php', data)['history']
except RuntimeError:
raise Exception("Failed to scrape case history")
@@ -1,0 +1,26 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>{% block title %}App{% endblock %}</title>
<link
rel="stylesheet"
href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css"
>
</head>
<body>
<main class="container">
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
<ul class="flashes">
{% for category, message in messages %}
<li class="{{ category }}">{{ message }}</li>
{% endfor %}
</ul>
{% endif %}
{% endwith %}
{% block content %}{% endblock %}
</main>
</body>
</html>
@@ -1,0 +1,34 @@
{% extends 'base.html' %}
{% block title %}Home{% endblock %}
{% block content %}
<h2>Welcome, {{ user.id }}!</h2>
<p>You are logged in.</p>
<a href="{{ url_for('main.logout') }}" role="button" class="secondary">Logout</a>
{% if user.admin == True %}
<details name="example" style="margin-top: 40px;">
<summary>Create a New User</summary>
<form method="post" action="{{ url_for('main.create_user') }}">
<input type="text" name="username" placeholder="New Username" required>
<input type="password" name="password" placeholder="New Password" required>
<button type="submit">Create User</button>
</form>
</details>
{% endif %}
<h3>Create a New Job</h3>
<form method="post" action="{{ url_for('main.enqueue_job') }}">
<input type="text" name="act" placeholder="Act Name*" required>
<input type="text" name="section" placeholder="Section">
<select name="state" id="state">
{% for code, name in states %}
<option value="{{ code }}">{{ name }}</option>
{% endfor %}
</select>
<button type="submit">Create User</button>
</form>
{% endblock %}
@@ -1,0 +1,16 @@
{% extends 'base.html' %}
{% block title %}Login{% endblock %}
{% block content %}
<h2>Login</h2>
{% if error %}
<article class="grid">
<p style="color: red">{{ error }}</p>
</article>
{% endif %}
<form method="post">
<input type="text" name="username" placeholder="Username" required>
<input type="password" name="password" placeholder="Password" required>
<button type="submit">Login</button>
</form>
{% endblock %}