301 lines
11 KiB
Python
301 lines
11 KiB
Python
import os
|
||
import sys
|
||
import urllib.request
|
||
import ssl
|
||
import re
|
||
import time
|
||
import json
|
||
from datetime import datetime
|
||
|
||
# Ensure terminal output supports UTF-8 on Windows
|
||
if sys.platform == 'win32':
|
||
try:
|
||
sys.stdout.reconfigure(encoding='utf-8')
|
||
except AttributeError:
|
||
pass
|
||
|
||
# Persian/Arabic to English digit mapping
|
||
FA_TO_EN = {
|
||
'۰': '0', '۱': '1', '۲': '2', '۳': '3', '۴': '4',
|
||
'۵': '5', '۶': '6', '۷': '7', '۸': '8', '۹': '9',
|
||
'٠': '0', '١': '1', '٢': '2', '٣': '3', '٤': '4',
|
||
'٥': '5', '٦': '6', '٧': '7', '٨': '8', '٩': '9'
|
||
}
|
||
|
||
def clean_persian_text(text):
|
||
"""Clean HTML tags and standardize spacing."""
|
||
if not text:
|
||
return ""
|
||
text = re.sub(r'<[^>]+>', '', text)
|
||
text = re.sub(r'\s+', ' ', text).strip()
|
||
return text
|
||
|
||
def to_english_digits(text):
|
||
"""Convert Persian/Arabic digits in a string to English digits."""
|
||
if not text:
|
||
return ""
|
||
return "".join(FA_TO_EN.get(char, char) for char in text)
|
||
|
||
def parse_numeric(text):
|
||
"""Convert a price string with commas and Persian digits into an integer."""
|
||
if not text:
|
||
return 0
|
||
clean_text = to_english_digits(text)
|
||
digits_only = re.sub(r'[^\d]', '', clean_text)
|
||
try:
|
||
return int(digits_only) if digits_only else 0
|
||
except ValueError:
|
||
return 0
|
||
|
||
def fetch_html_with_retry(url, retries=3, delay=2):
|
||
"""Fetch URL with retries, custom headers, and disabled SSL checks."""
|
||
ctx = ssl.create_default_context()
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
|
||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
|
||
'Accept-Language': 'fa,en-US;q=0.9,en;q=0.8',
|
||
'Cache-Control': 'no-cache',
|
||
'Pragma': 'no-cache'
|
||
}
|
||
|
||
# Make it a POST request by passing empty data (b"") to completely bypass CDN/ArvanCloud HTML caching
|
||
req = urllib.request.Request(url, data=b"", headers=headers)
|
||
|
||
for attempt in range(1, retries + 1):
|
||
try:
|
||
with urllib.request.urlopen(req, context=ctx, timeout=10) as response:
|
||
if response.status == 200:
|
||
html = response.read().decode('utf-8')
|
||
if "sekee" in html:
|
||
return html
|
||
else:
|
||
raise ValueError("Incorrect page content returned (missing expected coin tags).")
|
||
except Exception as e:
|
||
if attempt < retries:
|
||
time.sleep(delay)
|
||
delay *= 1.5
|
||
else:
|
||
print(f"⚠️ Connection error on attempt {attempt}: {e}")
|
||
return None
|
||
|
||
HISTORY_LOG = []
|
||
HISTORY_INITIALIZED = False
|
||
|
||
def initialize_history_from_excel():
|
||
global HISTORY_LOG, HISTORY_INITIALIZED
|
||
excel_filename = "coin_prices.xlsx"
|
||
if not HISTORY_INITIALIZED:
|
||
if os.path.exists(excel_filename):
|
||
try:
|
||
import pandas as pd
|
||
df = pd.read_excel(excel_filename, sheet_name="تاریخچه تغییرات")
|
||
HISTORY_LOG = df.to_dict(orient="records")
|
||
print(f"📥 Loaded {len(HISTORY_LOG)} history logs from existing '{excel_filename}'.")
|
||
except Exception as e:
|
||
print(f"💡 Initialized new history log in memory (failed to load: {e})")
|
||
HISTORY_INITIALIZED = True
|
||
|
||
def save_live_js_and_json(scraped_data):
|
||
"""Save scraped data to JSON and JS files for real-time HTML dashboard."""
|
||
current_time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
live_records = []
|
||
for row in scraped_data:
|
||
live_records.append({
|
||
"name": row["Name"],
|
||
"price_rial": row["Price_Rial"],
|
||
"price_toman": row["Price_Toman"],
|
||
"change": row["Change"],
|
||
"min_toman": row["Min_Price_Toman"],
|
||
"max_toman": row["Max_Price_Toman"],
|
||
"last_update": row["Last_Update"]
|
||
})
|
||
|
||
data_to_save = {
|
||
"last_fetch": current_time_str,
|
||
"prices": live_records
|
||
}
|
||
|
||
# Save as JSON
|
||
try:
|
||
with open("live_prices.json", "w", encoding="utf-8") as f:
|
||
json.dump(data_to_save, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"⚠️ Error saving live_prices.json: {e}")
|
||
|
||
# Save as JS (for file:// protocol CORS bypass)
|
||
try:
|
||
js_content = f"window.LIVE_COIN_PRICES = {json.dumps(data_to_save, ensure_ascii=False, indent=2)};"
|
||
with open("live_prices.js", "w", encoding="utf-8") as f:
|
||
f.write(js_content)
|
||
except Exception as e:
|
||
print(f"⚠️ Error saving live_prices.js: {e}")
|
||
|
||
def save_to_excel(scraped_data):
|
||
"""Save scraped data to Excel file with Latest Prices and History sheets."""
|
||
global HISTORY_LOG
|
||
excel_filename = "coin_prices.xlsx"
|
||
|
||
# Initialize history from Excel if we haven't already
|
||
initialize_history_from_excel()
|
||
|
||
# Create list for current scraped data
|
||
new_rows = []
|
||
current_time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
for row in scraped_data:
|
||
new_rows.append({
|
||
"زمان ثبت": current_time_str,
|
||
"نام سکه": row["Name"],
|
||
"قیمت فعلی (ریال)": row["Price_Rial"],
|
||
"قیمت فعلی (تومان)": row["Price_Toman"],
|
||
"میزان و درصد تغییر": row["Change"],
|
||
"کمترین قیمت روز (تومان)": row["Min_Price_Toman"],
|
||
"بیشترین قیمت روز (تومان)": row["Max_Price_Toman"],
|
||
"زمان بهروزرسانی سایت": row["Last_Update"]
|
||
})
|
||
|
||
import pandas as pd
|
||
df_new = pd.DataFrame(new_rows)
|
||
df_latest = df_new.drop(columns=["زمان ثبت"])
|
||
|
||
# Accumulate history in memory list
|
||
temp_history_log = HISTORY_LOG + new_rows
|
||
df_history = pd.DataFrame(temp_history_log)
|
||
|
||
# Write both sheets to Excel
|
||
try:
|
||
with pd.ExcelWriter(excel_filename, engine='openpyxl') as writer:
|
||
df_latest.to_excel(writer, sheet_name="قیمتهای لحظهای", index=False)
|
||
df_history.to_excel(writer, sheet_name="تاریخچه تغییرات", index=False)
|
||
|
||
# Write succeeded! Commit the temp log to our persistent memory HISTORY_LOG
|
||
HISTORY_LOG = temp_history_log
|
||
print(f"📊 Excel file updated: '{excel_filename}' (Latest & History sheets)")
|
||
return True
|
||
except Exception as e:
|
||
print(f"⚠️ Warning: Excel file is locked (likely open in Microsoft Excel).")
|
||
print(f"⚠️ {len(new_rows)} rows kept in memory and will auto-save on the next minute cycle when you close Excel. (Error: {e})")
|
||
return False
|
||
|
||
def run_scraper_cycle():
|
||
"""Runs a single cycle of the scraper."""
|
||
url = "https://www.tgju.org/coin"
|
||
html = fetch_html_with_retry(url)
|
||
|
||
if not html:
|
||
print("📡 Connection to TGJU failed. Retrying in next cycle...")
|
||
return False
|
||
|
||
market_rows = re.findall(r'<tr[^>]*data-market-row="([^"]+)"[^>]*>([\s\S]*?)<\/tr>', html)
|
||
|
||
target_keys = {
|
||
"sekee": "سکه امامی",
|
||
"sekeb": "سکه بهار آزادی",
|
||
"nim": "نیم سکه",
|
||
"rob": "ربع سکه",
|
||
"gerami": "سکه گرمی"
|
||
}
|
||
|
||
scraped_data = []
|
||
|
||
for row_name, row_content in market_rows:
|
||
if row_name in target_keys:
|
||
coin_display_name = target_keys[row_name]
|
||
tds = re.findall(r'<td[^>]*>([\s\S]*?)<\/td>', row_content)
|
||
|
||
if len(tds) >= 5:
|
||
raw_price_rial = clean_persian_text(tds[0])
|
||
raw_change = clean_persian_text(tds[1])
|
||
raw_min = clean_persian_text(tds[2])
|
||
raw_max = clean_persian_text(tds[3])
|
||
raw_time = clean_persian_text(tds[4])
|
||
|
||
price_rial = parse_numeric(raw_price_rial)
|
||
price_toman = price_rial // 10
|
||
|
||
min_rial = parse_numeric(raw_min)
|
||
min_toman = min_rial // 10
|
||
|
||
max_rial = parse_numeric(raw_max)
|
||
max_toman = max_rial // 10
|
||
|
||
clean_change = to_english_digits(raw_change)
|
||
clean_time = to_english_digits(raw_time)
|
||
|
||
scraped_data.append({
|
||
"Key": row_name,
|
||
"Name": coin_display_name,
|
||
"Price_Rial": price_rial,
|
||
"Price_Toman": price_toman,
|
||
"Change": clean_change,
|
||
"Min_Price_Toman": min_toman,
|
||
"Max_Price_Toman": max_toman,
|
||
"Last_Update": clean_time
|
||
})
|
||
|
||
if not scraped_data:
|
||
print("⚠️ No coin rows were matched in the HTML. Table structure may have changed.")
|
||
return False
|
||
|
||
# Save Excel
|
||
try:
|
||
save_to_excel(scraped_data)
|
||
except Exception as e:
|
||
print(f"⚠️ Error preparing Excel: {e}")
|
||
|
||
# Save JSON and JS for the dynamic HTML dashboard
|
||
try:
|
||
save_live_js_and_json(scraped_data)
|
||
except Exception as e:
|
||
print(f"⚠️ Error saving live assets: {e}")
|
||
|
||
# Print safely to console
|
||
print("\n" + "=" * 80)
|
||
print(f"{'نام سکه':<20} | {'قیمت (تومان)':<18} | {'تغییر':<18} | {'بهروزرسانی':<12}")
|
||
print("-" * 80)
|
||
for row in scraped_data:
|
||
name_ascii = row["Name"].ljust(20)
|
||
price_str = f"{row['Price_Toman']:,}"
|
||
change_str = row["Change"]
|
||
time_str = row["Last_Update"]
|
||
try:
|
||
print(f"{name_ascii} | {price_str:<18} | {change_str:<18} | {time_str:<12}")
|
||
except Exception:
|
||
print(f"{row['Key'].ljust(20)} | {price_str:<18} | {change_str:<18} | {time_str:<12}")
|
||
print("=" * 80 + "\n")
|
||
return True
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print(" 🪙 TGJU AUTOMATED COIN SCRAPER & SCHEDULER 🪙")
|
||
print(" 🔄 Running every 1 minute | Output: Excel & Realtime HTML")
|
||
print("=" * 60)
|
||
|
||
print(f"🚀 Initializing scraper at {datetime.now().strftime('%H:%M:%S')}...")
|
||
run_scraper_cycle()
|
||
|
||
while True:
|
||
try:
|
||
for remaining in range(60, 0, -1):
|
||
sys.stdout.write(f"\r⏳ Next fetch in {remaining:02d} seconds... ")
|
||
sys.stdout.flush()
|
||
time.sleep(1)
|
||
sys.stdout.write("\r📡 Fetching live prices... \n")
|
||
sys.stdout.flush()
|
||
|
||
run_scraper_cycle()
|
||
except KeyboardInterrupt:
|
||
print("\n👋 Scraper stopped by user. Goodbye!")
|
||
sys.exit(0)
|
||
except Exception as e:
|
||
print(f"\n⚠️ Unexpected error in loop: {e}")
|
||
print("⏳ Retrying in 10 seconds...")
|
||
time.sleep(10)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|