File "fetch_learninsta.py"

Full path: /home/algopkco/public_html/scraper/fetch_learninsta.py
File size: 7.32 B (7.32 KB bytes)
MIME-type: text/x-script.python
Charset: utf-8

Download   Open   Edit   Advanced Editor &nnbsp; Back

import json
import re
import html
import os
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests

H = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36',
}
URLS = r'D:\xampp\htdocs\algo\scraper\data\learninsta_urls.txt'
DATA = r'D:\xampp\htdocs\algo\scraper\data\learninsta_mcqs.jsonl'
PROG = r'D:\xampp\htdocs\algo\scraper\learninsta_progress.txt'
NTHREADS = 12

CHEM = re.compile(r'(?i)(chemical|reactions? and equations|acids?|bases?|salts?|carbon|organic|metals?|non-metals?|periodic|molecules?|matters? in our|is matter around|solutions?|compounds?|combustion|petroleum|coal|bonding|hydrocarbon|chemistry|separation|sorting materials|synthetic fibres|plastic|chemical substances)+')
PHYS = re.compile(r'(?i)(light|reflection|refraction|optics|lenses?|mirrors?|electricity|electric current|circuits?|magnet|motion|force|work and energy|power|sound|heat|temperature|gravitation|pressure|waves?|human eye|current|resistance|voltage|friction|momentum|inertia|density|buoyancy|thermometer|expansion|conduction|convection|radiation|kinetic|potential|universe|stars?|solar|floatation|physics|physical features)+')
BIO = re.compile(r'(?i)(cells?|tissues?|organs?|reproduction|reproductive|heredity|evolution|nutrition|respiration|photosynthesis|transportation|excretion|control and coordination|control \& coordination|hormones?|diseases?|health|immunity|microorganisms?|plants?|animals?|human|food|digestion|blood|heart|ecosystem|biosphere|biodiversity|classification|taxonomy|genetics|dna|biology|life processes|life cycle|vegetation|organism|pollination|seed|conservation)+')
SOC = re.compile(r'(?i)(social science|sst|history|geograph|political science|civics|economics?|constitution|janapada|mahajanapada|dynasty|empire|revolt|independence|movement|nationalism|colonial|democracy|governance|maratha|vijayanagara|medieval|ancient|modern india|gupta|maurya|delhi sultanate|mughal|rural|urban|village|panchayat|district|election|parliament|judiciary|culture|our country)+')
ENG = re.compile(r'(?i)(english|grammar|vocabulary|tense|preposition|conjunction|reading comprehension)+')
CS = re.compile(r'(?i)(computer|informatics|internet|programming|algorithm|python|coding|data science)+')
MATH = re.compile(r'(?i)(maths?|mathematics|algebra|geometry|trigonometry|calculus|statistics|probability|linear equations|quadratic|polynomials?|number system|integers|fractions?|decimals?|ratio|proportion|percentage|profit|interest|mensuration|triangles?|circles?|coordinate|real numbers|exponents|squares?|cubes?|sets?|functions?|graph|angles?|parallelogram|volume|surface area|pythagoras|playing with numbers|data handling|knowing our numbers|whole numbers)+')


def subj_of(title):
    if CS.search(title):
        return 'Computer Science General Test'
    if MATH.search(title) and 'science' not in title.lower():
        return 'Mathematics'
    if SOC.search(title):
        return 'Social-Studies-Mcqs'
    if ENG.search(title):
        return 'English Mcqs'
    if CHEM.search(title):
        return 'Chemistry Mcqs'
    if PHYS.search(title):
        return 'Physics Mcqs'
    if BIO.search(title):
        return 'Biology Mcqs'
    if re.search(r'(?i)science', title):
        return 'School Science'
    return 'School Science'


def parse_mcq_text(text):
    r = []
    blocks = re.split(r'Question\s+\d+\.\s*', text)
    for b in blocks:
        b = re.sub(r'\bAnswer\s+Answer\b', 'Answer', b)
        ans_letters = re.findall(r'Answer\s*:\s*\(?\s*([a-d])', b, re.I)
        if not ans_letters:
            continue
        clean = re.sub(r'Answer\s*:\s*\(?\s*[a-d]\)?.*$', '', b, flags=re.I)
        opts = re.findall(r'\(([a-d])\)\s*([^(].*?)(?=\s*\([a-d]\)\s|$)', clean, re.S)
        if len(opts) < 2:
            continue
        q = clean.split(' (a)')[0]
        q = re.sub(r'\s+', ' ', q).strip()
        q = q.lstrip('> ').strip()
        for cut in ['Online Test with Answers', 'MCQ Questions with Answers', ' with Answers', 'Answers ']:
            i = q.find(cut)
            if i != -1:
                q = q[i + len(cut):]
                break
        q = q.strip().lstrip('. :,').strip()
        if not q:
            continue
        letters = ['A', 'B', 'C', 'D']
        d = {l: '' for l in letters}
        for l, o in opts[:4]:
            d[l.upper()] = (l.upper() + ') ' + re.sub(r'\s+', ' ', o).strip()).strip()
        if not all(d.values()):
            continue
        corr = letters.index(ans_letters[0].upper()) + 1
        r.append({'question': q, 'options': [d['A'], d['B'], d['C'], d['D']], 'correct': corr})
    return r


def fetch_post(url):
    try:
        r = requests.get(url, headers=H, timeout=40)
        if r.status_code != 200:
            return url, 'http' + str(r.status_code), []
        h = r.text
        m = re.search(r'<title>(.*?)</title>', h, re.S)
        title = html.unescape(re.sub(r'<[^>]+>', '', m.group(1))).strip() if m else url
        title = re.sub(r'\s*\|\s*[^-]*$', '', title).strip()
        m = re.search(r'class="entry-content"', h)
        if not m:
            m = re.search(r'class="post-content"|class="td-post-content"|class="entry-summary"', h)
        body = h[m.start():] if m else h
        body = re.sub(r'<script.*?</script>|<style.*?</style>', '', body, flags=re.S)
        txt = html.unescape(re.sub(r'<[^>]+>', ' ', body))
        txt = re.sub(r'\s+', ' ', txt).strip()
        rows = parse_mcq_text(txt)
        subj = subj_of(title)
        out = []
        seen = set()
        for rw in rows:
            qn = rw['question']
            if qn in seen:
                continue
            seen.add(qn)
            out.append({'subject': subj, 'question': qn, 'options': rw['options'],
                        'correct': rw['correct'], 'topic': title, 'source': 'learninsta',
                        'url': url, 'explanation': ''})
        return url, title, out
    except Exception as e:
        return url, 'err ' + str(e), []


def main():
    urls = [u.strip() for u in open(URLS, encoding='utf-8') if u.strip()]
    urls = sorted(set(u for u in urls if 'mcq' in u.lower()))
    done = set()
    if os.path.exists(PROG):
        for ln in open(PROG, encoding='utf-8'):
            done.add(ln.strip())
    todo = [u for u in urls if u not in done]
    print(f'total={len(urls)} done={len(done)} todo={len(todo)}', flush=True)
    if not todo:
        return
    lock = threading.Lock()
    fw = open(DATA, 'a', encoding='utf-8')
    pg = open(PROG, 'a', encoding='utf-8')
    stats = {'ok': 0, 'err': 0, 'rows': 0}
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=NTHREADS) as ex:
        futs = {ex.submit(fetch_post, u): u for u in todo}
        for i, f in enumerate(as_completed(futs), 1):
            u, info, rows = f.result()
            with lock:
                if rows:
                    for rw in rows:
                        fw.write(json.dumps(rw, ensure_ascii=False) + '\n')
                    stats['ok'] += 1
                    stats['rows'] += len(rows)
                else:
                    stats['err'] += 1
                pg.write(u + '\n')
                if i % 25 == 0:
                    print(f'[{i}/{len(todo)}] rows={stats["rows"]} ok={stats["ok"]} err={stats["err"]} elapsed={time.time()-t0:.0f}s', flush=True)
    fw.close()
    pg.close()
    print('DONE', stats, flush=True)


if __name__ == '__main__':
    main()