welcome.py fixed to greet only new users. timezone plugin added.
This commit is contained in:
@@ -0,0 +1,192 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Time Zone Plugin – completely hardcoded-free using Open-Meteo APIs.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import aiohttp
|
||||
import simplematrixbotlib as botlib
|
||||
from urllib.parse import quote
|
||||
from datetime import datetime
|
||||
|
||||
def format_ampm(dt_str: str) -> str:
|
||||
"""Convert ISO datetime to AM/PM format."""
|
||||
try:
|
||||
if '+' in dt_str:
|
||||
dt_str = dt_str.split('+')[0]
|
||||
if '.' in dt_str:
|
||||
dt_str = dt_str.split('.')[0]
|
||||
dt_str = dt_str.replace('T', ' ')
|
||||
dt = datetime.fromisoformat(dt_str)
|
||||
return dt.strftime("%I:%M:%S %p").lstrip("0")
|
||||
except:
|
||||
return dt_str
|
||||
|
||||
async def geocode_city(session: aiohttp.ClientSession, city: str) -> tuple[float, float, str] | None:
|
||||
"""
|
||||
Open-Meteo Geocoding API (free, no key, no hardcoding).
|
||||
Returns (latitude, longitude, display_name) or None.
|
||||
"""
|
||||
url = f"https://geocoding-api.open-meteo.com/v1/search?name={quote(city)}&count=1&language=en&format=json"
|
||||
|
||||
try:
|
||||
async with session.get(url, timeout=10) as resp:
|
||||
if resp.status == 200:
|
||||
data = await resp.json()
|
||||
if data.get("results") and len(data["results"]) > 0:
|
||||
result = data["results"][0]
|
||||
lat = result["latitude"]
|
||||
lon = result["longitude"]
|
||||
name = result.get("name", city)
|
||||
country = result.get("country", "")
|
||||
admin1 = result.get("admin1", "")
|
||||
|
||||
# Build display name: "Lahore, Punjab, Pakistan"
|
||||
display_parts = [name]
|
||||
if admin1 and admin1 != name:
|
||||
display_parts.append(admin1)
|
||||
if country:
|
||||
display_parts.append(country)
|
||||
display_name = ", ".join(display_parts)
|
||||
|
||||
logging.info(f"Geocoded: {city} → {display_name} ({lat}, {lon})")
|
||||
return lat, lon, display_name
|
||||
else:
|
||||
logging.warning(f"Geocoding API HTTP {resp.status} for {city}")
|
||||
except Exception as e:
|
||||
logging.warning(f"Geocoding error: {e}")
|
||||
return None
|
||||
|
||||
async def get_timezone(lat: float, lon: float) -> str | None:
|
||||
"""
|
||||
Get timezone name from coordinates using timezonedb (free tier, no key).
|
||||
Alternative: use Open-Meteo's time API directly.
|
||||
"""
|
||||
# Open-Meteo's time API accepts coordinates directly
|
||||
# We'll use this instead of timezonedb
|
||||
return None # Will be handled in fetch_time_by_coords
|
||||
|
||||
async def fetch_time_by_coords(session: aiohttp.ClientSession, lat: float, lon: float) -> dict | None:
|
||||
"""
|
||||
Get current time using Open-Meteo (no key required).
|
||||
"""
|
||||
url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t_weather=true&timezone=auto&timeformat=unixtime"
|
||||
|
||||
try:
|
||||
async with session.get(url, timeout=10) as resp:
|
||||
if resp.status == 200:
|
||||
data = await resp.json()
|
||||
current = data.get("current_weather", {})
|
||||
timezone = data.get("timezone", "Unknown")
|
||||
unixtime = current.get("time")
|
||||
temperature = current.get("temperature")
|
||||
|
||||
if unixtime:
|
||||
# Convert UNIX timestamp to datetime
|
||||
dt = datetime.fromtimestamp(unixtime)
|
||||
return {
|
||||
"datetime": dt.isoformat(),
|
||||
"timezone": timezone,
|
||||
"temperature": temperature
|
||||
}
|
||||
except Exception as e:
|
||||
logging.warning(f"Time fetch error: {e}")
|
||||
return None
|
||||
|
||||
async def fetch_time_by_zone(session: aiohttp.ClientSession, zone: str) -> dict | None:
|
||||
"""Get current time for a named timezone using Open-Meteo."""
|
||||
# Open-Meteo doesn't have named timezone endpoint, need to geocode a representative city
|
||||
# Fallback to worldtimeapi.org for IANA zones
|
||||
url = f"http://worldtimeapi.org/api/timezone/{zone}"
|
||||
try:
|
||||
async with session.get(url, timeout=10) as resp:
|
||||
if resp.status == 200:
|
||||
return await resp.json()
|
||||
except Exception as e:
|
||||
logging.warning(f"Timezone API error: {e}")
|
||||
return None
|
||||
|
||||
async def resolve_time(session: aiohttp.ClientSession, query: str) -> tuple[dict | None, str]:
|
||||
"""Main resolution: geocode any city, then get time."""
|
||||
query = query.strip().lower()
|
||||
|
||||
# Check if it's an IANA zone (contains '/')
|
||||
if '/' in query or query in ("utc", "gmt"):
|
||||
data = await fetch_time_by_zone(session, query)
|
||||
if data:
|
||||
return data, query.upper()
|
||||
return None, f"Timezone '{query}' not found"
|
||||
|
||||
# Geocode the city (no hardcoding!)
|
||||
geocode_result = await geocode_city(session, query)
|
||||
if not geocode_result:
|
||||
return None, f"Could not find city '{query}'. Try being more specific."
|
||||
|
||||
lat, lon, display_name = geocode_result
|
||||
|
||||
# Get time from coordinates
|
||||
data = await fetch_time_by_coords(session, lat, lon)
|
||||
if not data:
|
||||
return None, f"Could not get time for '{display_name}'"
|
||||
|
||||
return data, display_name
|
||||
|
||||
def format_response(data: dict, display_name: str) -> str:
|
||||
"""Format time data into HTML."""
|
||||
raw_time = data.get("datetime", "")
|
||||
local_time = format_ampm(raw_time) if raw_time else "Unknown"
|
||||
tz = data.get("timezone", "Unknown")
|
||||
temp = data.get("temperature")
|
||||
temp_str = f"<br>🌡️ <strong>Temperature:</strong> {temp}°C" if temp is not None else ""
|
||||
|
||||
return f"""
|
||||
<details>
|
||||
<summary><strong>🕒 Time in {display_name}</strong></summary>
|
||||
<p>
|
||||
📍 <strong>Timezone:</strong> {tz}<br>
|
||||
📅 <strong>Local time:</strong> {local_time}{temp_str}
|
||||
</p>
|
||||
</details>
|
||||
"""
|
||||
|
||||
def help_text() -> str:
|
||||
return """
|
||||
<details>
|
||||
<summary><strong>🕒 Time Plugin Help</strong></summary>
|
||||
<p>
|
||||
<strong>!time <any city></strong> – Get current time for ANY city worldwide<br>
|
||||
<strong>!time <IANA zone></strong> – e.g., Europe/London, Asia/Karachi<br>
|
||||
<strong>!time help</strong> – Show this help<br><br>
|
||||
<strong>Examples:</strong><br>
|
||||
<code>!time Lahore</code><br>
|
||||
<code>!time New York</code><br>
|
||||
<code>!time Paris</code><br>
|
||||
<code>!time Asia/Karachi</code><br><br>
|
||||
<em>No city names are hardcoded. The bot uses Open-Meteo's geocoding API.</em>
|
||||
</p>
|
||||
</details>
|
||||
"""
|
||||
|
||||
def setup(bot):
|
||||
logging.info("Time plugin (zero hardcoded cities) loaded.")
|
||||
|
||||
async def handle_command(room, message, bot, prefix, config):
|
||||
match = botlib.MessageMatch(room, message, bot, prefix)
|
||||
if not (match.is_not_from_this_bot() and match.prefix() and match.command("time")):
|
||||
return
|
||||
|
||||
args = match.args()
|
||||
if not args or args[0].lower() == "help":
|
||||
await bot.api.send_markdown_message(room.room_id, help_text())
|
||||
return
|
||||
|
||||
query = " ".join(args).strip()
|
||||
await bot.api.send_text_message(room.room_id, f"🕒 Looking up time for: {query}...")
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
data, display = await resolve_time(session, query)
|
||||
if data is None:
|
||||
await bot.api.send_text_message(room.room_id, f"❌ {display}")
|
||||
return
|
||||
await bot.api.send_markdown_message(room.room_id, format_response(data, display))
|
||||
logging.info(f"Time sent for {query}")
|
||||
Reference in New Issue
Block a user