127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
import os
|
|
import json
|
|
import requests
|
|
from jdatetime import datetime, timedelta
|
|
from dotenv import load_dotenv
|
|
|
|
# Load local configuration without overriding variables supplied by the shell or CI.
|
|
load_dotenv()
|
|
|
|
BARGHEMAN_TOKEN = os.getenv("BARGHEMAN_TOKEN", "").strip()
|
|
EITAAYAR_TOKEN = os.getenv("EITAAYAR_TOKEN", "").strip()
|
|
CHAT_ID = os.getenv("CHAT_ID", "").strip()
|
|
BILL_IDS = [bill_id.strip() for bill_id in os.getenv("BILL_IDS", "").split(",") if bill_id.strip()]
|
|
BLACKOUTS_FILE = os.getenv("BLACKOUTS_FILE", "").strip()
|
|
|
|
required_settings = {
|
|
"BARGHEMAN_TOKEN": BARGHEMAN_TOKEN,
|
|
"EITAAYAR_TOKEN": EITAAYAR_TOKEN,
|
|
"CHAT_ID": CHAT_ID,
|
|
"BILL_IDS": BILL_IDS,
|
|
"BLACKOUTS_FILE": BLACKOUTS_FILE,
|
|
}
|
|
missing_settings = [name for name, value in required_settings.items() if not value]
|
|
if missing_settings:
|
|
raise EnvironmentError(f"Missing required environment variables: {', '.join(missing_settings)}")
|
|
|
|
today = datetime.now()
|
|
to_date = today + timedelta(days=5)
|
|
|
|
|
|
def get_blackouts(bill_id) -> list:
|
|
url = "https://uiapi2.saapa.ir/api/ebills/PlannedBlackoutsReport"
|
|
headers = {
|
|
'accept': 'application/json',
|
|
"Authorization": f"Bearer {BARGHEMAN_TOKEN}",
|
|
}
|
|
payload = {
|
|
"bill_id": bill_id,
|
|
"from_date": today.strftime("%Y/%m/%d"),
|
|
"to_date": to_date.strftime("%Y/%m/%d"),
|
|
}
|
|
response = requests.post(url, headers=headers, json=payload)
|
|
if response.ok:
|
|
return response.json().get("data", [])
|
|
else:
|
|
print(f"❌ Failed to get blackouts: {response.text}")
|
|
return []
|
|
|
|
|
|
def load_blackouts() -> dict:
|
|
if os.path.exists(BLACKOUTS_FILE):
|
|
with open(BLACKOUTS_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
|
|
def save_blackouts(data) -> None:
|
|
with open(BLACKOUTS_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def format_message(bill_id, entry):
|
|
RTLM = "\u200f"
|
|
return f"{RTLM}\n{RTLM}".join([
|
|
f"⚡️ اطلاعیه خاموشی برق",
|
|
f"🆔 شناسه قبض: {bill_id}",
|
|
f"📅 تاریخ: {entry['outage_date']}",
|
|
f"⏰ ساعت: {entry['outage_start_time']} تا {entry['outage_stop_time']}",
|
|
f"📍 محل: {entry['outage_address']}",
|
|
f"📝 دلیل: {entry['reason_outage']}",
|
|
f"📥 ارسال شده در {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
|
f".",
|
|
])
|
|
|
|
|
|
def send_message(text):
|
|
url = f"https://eitaayar.ir/api/{EITAAYAR_TOKEN}/sendMessage"
|
|
payload = {
|
|
"chat_id": CHAT_ID,
|
|
"text": text,
|
|
"parse_mode": "HTML"
|
|
}
|
|
response = requests.get(url, params=payload)
|
|
if not response.ok:
|
|
print(f"❌ Failed to send message: {response.text}")
|
|
|
|
|
|
def is_entries_valid(entries: list) -> bool:
|
|
return 0 < len(entries) < 10
|
|
|
|
|
|
def main() -> None:
|
|
previous_data = load_blackouts()
|
|
updated_data = previous_data.copy()
|
|
|
|
for bill_id in BILL_IDS:
|
|
new_blackouts = get_blackouts(bill_id)
|
|
old_blackouts = previous_data.get(bill_id, [])
|
|
|
|
new_entries = [entry for entry in new_blackouts if entry not in old_blackouts]
|
|
|
|
if not is_entries_valid(new_entries):
|
|
print('unusual entries was received. count:', len(new_entries))
|
|
print(json.dumps(new_entries, ensure_ascii=False, indent=2))
|
|
continue
|
|
|
|
for entry in new_entries:
|
|
message = format_message(bill_id, entry)
|
|
send_message(message)
|
|
|
|
if new_entries:
|
|
updated_data[bill_id] = old_blackouts + new_entries
|
|
|
|
save_blackouts(updated_data)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|
|
"""
|
|
curl ^"https://uiapi2.saapa.ir/api/ebills/PlannedBlackoutsReport^" ^
|
|
--compressed ^
|
|
-X POST ^
|
|
-H ^"Authorization: Bearer ...^" ^
|
|
--data-raw ^"{\"bill_id\":\"7619314604429\",\"from_date\":\"1404/03/11\",\"to_date\":\"1404/03/16\"}^"
|
|
"""
|