jakub/weatherbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

openmeteo

Browse Source
This commit is contained in:
Jakub Žáček
2026-02-01 17:32:16 +01:00
parent 901fd79cb6
commit 4afa35bce5
1 changed files with 111 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
main.py
View File

@@ -12,12 +12,19 @@ SMS_PASS = os.getenv("SMS_PASS", "Cupakabra17")
RECEIVE_URL = "https://sms.internet-master.cz/receive/"
SEND_URL = "https://sms.internet-master.cz/send/"
# Latency target: worst-case <= ~5s
POLL_INTERVAL = 1 # seconds
# ===== TIMING =====
POLL_INTERVAL = 1 # seconds (low latency)
# Persistence
# ===== STATE =====
STATE_FILE = Path("weatherbot_state.json")
MAX_SEEN_OFFSETS = 5000 # cap memory/state growth
MAX_SEEN_OFFSETS = 5000
# ===== OPEN-METEO =====
GEO_URL = "https://geocoding-api.open-meteo.com/v1/search"
WEATHER_URL = "https://api.open-meteo.com/v1/forecast"
# city -> (lat, lon)
GEO_CACHE = {}
HEADERS = {
"User-Agent": "weatherbot/1.0 (jakub@jimbuntu)"
@@ -26,6 +33,10 @@ HEADERS = {
session = requests.Session()
session.headers.update(HEADERS)
# =========================================================
# SMS API
# =========================================================
def fetch_inbox(limit=5):
r = session.get(
RECEIVE_URL,
@@ -53,35 +64,96 @@ def send_sms(number, message):
r.raise_for_status()
print("📤 SEND:", r.text)
# =========================================================
# WEATHER (Open-Meteo, no registration)
# =========================================================
CODE_MAP = {
0: "Clear",
1: "Mostly clear",
2: "Partly cloudy",
3: "Cloudy",
45: "Fog",
48: "Fog",
51: "Drizzle",
53: "Drizzle",
55: "Drizzle",
61: "Rain",
63: "Rain",
65: "Heavy rain",
71: "Snow",
73: "Snow",
75: "Heavy snow",
80: "Showers",
81: "Showers",
82: "Heavy showers",
95: "Storm"
}
def get_weather(city):
city = city.strip()
if not city:
return "❌ Send a city name."
try:
key = city.lower()
# --- GEO ---
if key in GEO_CACHE:
lat, lon = GEO_CACHE[key]
else:
r = session.get(
GEO_URL,
params={"name": city, "count": 1},
timeout=2
)
r.raise_for_status()
data = r.json()
results = data.get("results")
if not results:
return f"❌ Unknown city: {city}"
lat = results[0]["latitude"]
lon = results[0]["longitude"]
GEO_CACHE[key] = (lat, lon)
# --- WEATHER ---
r = session.get(
f"https://wttr.in/{city}?format=j1",
timeout=4 # keep it under 5s total budget
WEATHER_URL,
params={
"latitude": lat,
"longitude": lon,
"daily": "temperature_2m_mean,weathercode",
"timezone": "auto"
},
timeout=2
)
except requests.exceptions.RequestException:
return f"⚠️ Weather service timeout for {city}. Try again later."
r.raise_for_status()
data = r.json()
if r.status_code != 200:
return f"❌ Weather unavailable for {city}"
daily = data.get("daily", {})
dates = daily.get("time", [])[:3]
temps = daily.get("temperature_2m_mean", [])[:3]
codes = daily.get("weathercode", [])[:3]
data = r.json()
weather = data.get("weather", [])
if not dates:
return f"❌ Weather unavailable for {city}"
if len(weather) < 3:
return f"❌ Weather unavailable for {city}"
msg = f"🌦 {city}:\n"
for d, t, c in zip(dates, temps, codes):
wd = datetime.strptime(d, "%Y-%m-%d").strftime("%a")
desc = CODE_MAP.get(c, "Weather")
msg += f"{wd}: {round(t)}°C {desc}\n"
msg = f"🌦 Weather for {city}:\n"
for day in weather[:3]:
date = day["date"]
avgtemp = day["avgtempC"]
desc = day["hourly"][0]["weatherDesc"][0]["value"]
weekday = datetime.strptime(date, "%Y-%m-%d").strftime("%a")
msg += f"{weekday}: {avgtemp}°C, {desc}\n"
return msg.strip()
return msg.strip()
except Exception:
return f"⚠ Weather service error for {city}"
# =========================================================
# STATE HANDLING
# =========================================================
def load_state():
if not STATE_FILE.exists():
@@ -89,102 +161,59 @@ def load_state():
try:
with STATE_FILE.open("r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
return {"seen_offsets": []}
if "seen_offsets" not in data or not isinstance(data["seen_offsets"], list):
data["seen_offsets"] = []
return data
return data if isinstance(data, dict) else {"seen_offsets": []}
except Exception:
# If state is corrupt, start clean (but startup priming still prevents backlog replies)
return {"seen_offsets": []}
def save_state(state):
# Keep only the newest N offsets (as ints if possible)
seen = state.get("seen_offsets", [])
if len(seen) > MAX_SEEN_OFFSETS:
seen = seen[-MAX_SEEN_OFFSETS:]
state["seen_offsets"] = seen
state["seen_offsets"] = seen[-MAX_SEEN_OFFSETS:]
tmp = STATE_FILE.with_suffix(".json.tmp")
tmp = STATE_FILE.with_suffix(".tmp")
with tmp.open("w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False)
tmp.replace(STATE_FILE)
def prime_seen_offsets(seen_set):
"""
On startup, mark whatever is currently in the inbox as 'already seen'
so we do NOT reply to backlog after a restart.
"""
try:
data = fetch_inbox(limit=20)
inbox = data.get("inbox", [])
primed = 0
for sms in inbox:
for sms in data.get("inbox", []):
offset = sms.get("offset")
if offset is None:
continue
if offset not in seen_set:
if offset is not None:
seen_set.add(offset)
primed += 1
if primed:
print(f"🧹 Primed {primed} existing message(s) as already-seen (no backlog replies).")
else:
print("🧹 No existing messages to prime.")
print("🧹 Inbox primed (no backlog replies).")
except Exception as e:
print("⚠ Prime failed (continuing):", e)
print("⚠ Prime failed:", e)
def add_seen(seen_set, seen_list, offset):
# offset can be int/str depending on API; store exactly as returned
if offset in seen_set:
return False
seen_set.add(offset)
seen_list.append(offset)
# cap list size
if len(seen_list) > MAX_SEEN_OFFSETS:
# drop oldest chunk
drop = len(seen_list) - MAX_SEEN_OFFSETS
for old in seen_list[:drop]:
seen_set.discard(old)
del seen_list[:drop]
return True
# =========================================================
# MAIN LOOP
# =========================================================
# ===== MAIN =====
print("📡 SMS Weather Bot started (≤5s reply + persistent offsets)")
print("📡 SMS Weather Bot started (Open-Meteo, no registration)")
state = load_state()
seen_list = state.get("seen_offsets", [])
seen_set = set(seen_list)
# Prevent replying to old messages sitting in inbox after restart
prime_seen_offsets(seen_set)
# Sync list to set after priming (preserve order by appending primed ones)
# (We don't know which were primed vs existing; easiest is to rewrite list from set as a stable list)
# But to preserve bounded size, we just extend list with items missing:
for off in list(seen_set):
if off not in seen_list:
seen_list.append(off)
state["seen_offsets"] = seen_list
state["seen_offsets"] = list(seen_set)
save_state(state)
while True:
try:
data = fetch_inbox(limit=5)
inbox = data.get("inbox", [])
for sms in inbox:
for sms in data.get("inbox", []):
offset = sms.get("offset")
if offset is None:
continue
# If already processed, skip
if offset in seen_set:
if offset in seen_set or offset is None:
continue
phone = sms.get("phone", "")
text = (sms.get("message") or "").strip()
# Mark seen immediately to avoid duplicate replies if send fails mid-loop
add_seen(seen_set, seen_list, offset)
seen_set.add(offset)
seen_list.append(offset)
state["seen_offsets"] = seen_list
save_state(state)
@@ -197,6 +226,5 @@ while True:
except Exception as e:
print("⚠ ERROR:", e)
# Keep latency target: no long backoff; next poll in 1s
time.sleep(POLL_INTERVAL)