212 lines
7.1 KiB
Python
212 lines
7.1 KiB
Python
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
|
|
import aiohttp
|
|
from bs4 import BeautifulSoup
|
|
from colorama import Fore, Style, init
|
|
|
|
init(autoreset=True)
|
|
|
|
BASE_URL = "https://nepalicalendar.rat32.com/index_nep.php"
|
|
DATA_DIR = "./data"
|
|
|
|
|
|
# Helper: wait between requests
|
|
async def delay(ms):
|
|
await asyncio.sleep(ms / 1000)
|
|
|
|
|
|
async def fetch_month(session, year, month, save_dir=True, attempt=1):
|
|
month_log = f"[{year}/{month}]"
|
|
file_path = f"{DATA_DIR}/{year}/{month}.json"
|
|
|
|
print(Fore.CYAN + f"\nFetching data for {month_log} (Attempt {attempt})")
|
|
|
|
if os.path.exists(file_path):
|
|
print(Fore.LIGHTBLACK_EX + "Skipped: File already exists.")
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
try:
|
|
print(Fore.BLUE + "Sending request...")
|
|
payload = {
|
|
"selYear": year,
|
|
"selMonth": month,
|
|
"viewCalander": "View Calander",
|
|
}
|
|
|
|
async with session.post(BASE_URL, data=payload, timeout=15) as resp:
|
|
if resp.status != 200:
|
|
raise Exception(f"HTTP {resp.status}")
|
|
html = await resp.text()
|
|
|
|
print(Fore.GREEN + "Page fetched successfully. Parsing HTML...")
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
|
|
# Metadata
|
|
metadata = {
|
|
"en": soup.select_one("#entarikYr").get_text(strip=True)
|
|
if soup.select_one("#entarikYr")
|
|
else "",
|
|
"np": soup.select_one("#yren").get_text(strip=True)
|
|
if soup.select_one("#yren")
|
|
else "",
|
|
}
|
|
print(Fore.LIGHTBLACK_EX + f"Metadata: {metadata['np']} ({metadata['en']})")
|
|
|
|
# Helper to extract lists
|
|
def extract_list(selector, remove_tags=None):
|
|
el = soup.select_one(selector)
|
|
if not el:
|
|
return []
|
|
remove_tags = remove_tags or []
|
|
for tag in remove_tags:
|
|
t = el.select_one(tag)
|
|
if t:
|
|
t.decompose()
|
|
return [line.strip() for line in el.get_text().split("\n") if line.strip()]
|
|
|
|
holi_fest = extract_list("#holi", ["b", "a"])
|
|
marriage = extract_list("#bibah", ["b"])
|
|
bratabandha = extract_list("#bratabandha", ["b"])
|
|
|
|
print(Fore.MAGENTA + f"Holidays/Festivals: {len(holi_fest)}")
|
|
print(Fore.MAGENTA + f"Marriage Dates: {len(marriage)}")
|
|
print(Fore.MAGENTA + f"Bratabandha Dates: {len(bratabandha)}")
|
|
|
|
# Extract days
|
|
days = []
|
|
cells = soup.select(".cells")
|
|
for i, cell in enumerate(cells):
|
|
cell_soup = BeautifulSoup(cell.decode_contents(), "html.parser")
|
|
|
|
n = (
|
|
cell_soup.select_one("#nday").get_text(strip=True)
|
|
if cell_soup.select_one("#nday")
|
|
else ""
|
|
)
|
|
e = (
|
|
cell_soup.select_one("#eday").get_text(strip=True)
|
|
if cell_soup.select_one("#eday")
|
|
else ""
|
|
)
|
|
t = (
|
|
cell_soup.select_one("#dashi").get_text(strip=True)
|
|
if cell_soup.select_one("#dashi")
|
|
else ""
|
|
)
|
|
f = (
|
|
cell_soup.select_one("#fest").get_text(strip=True)
|
|
if cell_soup.select_one("#fest")
|
|
else ""
|
|
)
|
|
color = ""
|
|
font_tag = cell_soup.select_one("#nday font")
|
|
if font_tag and font_tag.has_attr("color"):
|
|
color = font_tag["color"].lower()
|
|
h = color == "red"
|
|
|
|
days.append({"d": (i % 7) + 1, "n": n, "e": e, "t": t, "f": f, "h": h})
|
|
|
|
print(Fore.GREEN + f"Parsed {len(days)} days successfully.")
|
|
|
|
data = {
|
|
"metadata": metadata,
|
|
"days": days,
|
|
"holiFest": holi_fest,
|
|
"marriage": marriage,
|
|
"bratabandha": bratabandha,
|
|
}
|
|
|
|
if save_dir:
|
|
os.makedirs(f"{DATA_DIR}/{year}", exist_ok=True)
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
json.dump(
|
|
data,
|
|
f,
|
|
indent=2,
|
|
ensure_ascii=False,
|
|
)
|
|
print(Fore.LIGHTBLACK_EX + f"Saved individual month: {file_path}")
|
|
|
|
return data
|
|
|
|
except Exception as err:
|
|
print(Fore.RED + f"Error fetching {month_log}: {err}")
|
|
if attempt < 3:
|
|
wait_time = 3000 * attempt
|
|
print(Fore.YELLOW + f"Retrying in {wait_time / 1000}s...")
|
|
await delay(wait_time)
|
|
return await fetch_month(session, year, month, save_dir, attempt + 1)
|
|
else:
|
|
print(Fore.RED + f"Failed after 3 attempts for {month_log}")
|
|
return None
|
|
|
|
|
|
async def scrape_year(session, year, save_single=True, save_dir=True):
|
|
print(Style.BRIGHT + f"\nStarting scrape for year: {Fore.CYAN}{year}")
|
|
print(f" Mode: {'Single JSON' if save_single else ''} {'Directory' if save_dir else ''}\n")
|
|
year_data = {}
|
|
success_count = 0
|
|
|
|
for month in range(1, 13):
|
|
data = await fetch_month(session, year, month, save_dir=save_dir)
|
|
if data:
|
|
year_data[str(month)] = data
|
|
success_count += 1
|
|
else:
|
|
print(Fore.RED + f"Failed to gather data for month {month}")
|
|
|
|
await delay(2000) # delay between months
|
|
|
|
if success_count == 12 and save_single:
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
year_file_path = f"{DATA_DIR}/{year}.json"
|
|
with open(year_file_path, "w", encoding="utf-8") as f:
|
|
json.dump(year_data, f, indent=2, ensure_ascii=False)
|
|
print(Style.BRIGHT + Fore.GREEN + f"\nSaved aggregated file: {year_file_path}")
|
|
|
|
fail_count = 12 - success_count
|
|
print(Fore.LIGHTBLACK_EX + f"Success: {success_count} Failed: {fail_count}\n")
|
|
|
|
|
|
async def scrape_years(start_year, end_year, save_single=True, save_dir=True):
|
|
async with aiohttp.ClientSession() as session:
|
|
for year in range(start_year, end_year + 1):
|
|
await scrape_year(session, year, save_single, save_dir)
|
|
if year < end_year:
|
|
print(Fore.YELLOW + "Waiting 5 seconds before next year...")
|
|
await delay(5000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Nepali Calendar Scraper")
|
|
parser.add_argument("years", nargs="+", type=int, help="Year or year range (start end)")
|
|
parser.add_argument("--single", nargs="?", const="json", help="Generate single JSON (e.g. --single json)")
|
|
parser.add_argument("--dir", nargs="?", const="format", help="Generate directory format (e.g. --dir format)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Determine output mode
|
|
save_single = False
|
|
save_dir = False
|
|
|
|
if not args.single and not args.dir:
|
|
# Default behavior: both
|
|
save_single = True
|
|
save_dir = True
|
|
else:
|
|
if args.single:
|
|
save_single = True
|
|
if args.dir:
|
|
save_dir = True
|
|
|
|
if len(args.years) == 1:
|
|
start_year = end_year = args.years[0]
|
|
else:
|
|
start_year, end_year = args.years[0], args.years[1]
|
|
|
|
asyncio.run(scrape_years(start_year, end_year, save_single, save_dir))
|