✨ 支持地理位置解析

config.gen.ini

@@ -23,3 +23,6 @@ access_token_secret = ABCD
 admin = 0
 lofter_channel = 0
 lofter_channel_username = username
+
+[api]
+amap_key = ABCD

defs/glover.py

@@ -16,6 +16,8 @@ access_token_secret: str = ""
 admin: int = 0
 lofter_channel: int = 0
 lofter_channel_username: str = ""
+# [api]
+amap_key: str = ""
 
 config = RawConfigParser()
 config.read("config.ini")
@@ -29,6 +31,7 @@ access_token_secret = config.get("twitter", "access_token_secret", fallback=acce
 admin = config.getint("post", "admin", fallback=admin)
 lofter_channel = config.getint("post", "lofter_channel", fallback=lofter_channel)
 lofter_channel_username = config.get("post", "lofter_channel_username", fallback=lofter_channel_username)
+amap_key = config.get("api", "amap_key", fallback=amap_key)
 try:
     ipv6 = strtobool(ipv6)
 except ValueError:

modules/geo.py

@@ -0,0 +1,43 @@
+import contextlib
+
+from urllib.parse import quote
+
+from pyrogram import Client, filters
+from pyrogram.types import Message
+
+from defs.glover import amap_key
+
+from init import user_me, request
+
+REQUEST_URL = f"https://restapi.amap.com/v3/geocode/geo?key={amap_key}&"
+
+
+@Client.on_message(filters.incoming &
+                   filters.command(["geo", f"geo@{user_me.username}"]))
+async def geo_command(_: Client, message: Message):
+    if len(message.command) <= 1:
+        await message.reply('没有找到要查询的中国 经纬度/地址 ...')
+        return
+    mode, lat, lon = "address", 0, 0  # noqa
+    with contextlib.suppress(ValueError, IndexError):
+        lat, lon = float(message.command[1]), float(message.command[2])
+        mode = "location"
+    if mode == "location":
+        try:
+            geo = (await request.get(f"{REQUEST_URL.replace('/geo?', '/regeo?')}location={lat},{lon}")).json()
+            formatted_address = geo["regeocode"]["formatted_address"]
+            assert isinstance(formatted_address, str)
+        except (KeyError, AssertionError):
+            await message.reply(f"无法解析经纬度 {lat}, {lon}", quote=True)
+            return
+    else:
+        try:
+            geo = (await request.get(f"{REQUEST_URL}address={quote(' '.join(message.command[1:]).strip())}")).json()
+            formatted_address = geo["geocodes"][0]["formatted_address"]
+            lat, lon = geo["geocodes"][0]["location"].split(",")
+        except (KeyError, IndexError, ValueError):
+            await message.reply(f"无法解析地址 {' '.join(message.command[1:]).strip()}", quote=True)
+            return
+    msg = await message.reply_location(longitude=float(lat), latitude=float(lon), quote=True)
+    await msg.reply(f"坐标：`{lat},{lon}`\n"
+                    f"地址：<b>{formatted_address}</b>", quote=True)

modules/lofter.py

@@ -28,9 +28,7 @@ async def lofter_share(_: Client, message: Message):
                     await img[0].reply_to(message, static=static)
                 else:
                     await message.reply_media_group(media=await input_media(img[:9], static), quote=True)
-            elif "front/blog" in url:
-                pass
-            else:
+            elif "front/blog" not in url:
                 text, avatar, username, status_link = await get_loft_user(url)
                 if avatar:
                     await message.reply_photo(

modules/twitter_api.py

@@ -29,13 +29,19 @@ async def twitter_share(client: Client, message: Message):
             if url.hostname and url.hostname in ["twitter.com", "vxtwitter.com"]:
                 if url.path.find('status') >= 0:
                     status_id = str(url.path[url.path.find('status') + 7:].split("/")[0]).split("?")[0]
+                    url_json = None
                     with ThreadPoolExecutor() as executor:
+                        for _ in range(3):
                             try:
                                 future = client.loop.run_in_executor(executor, twitter_api.GetStatus, status_id)
                                 url_json = await asyncio.wait_for(future, timeout=30, loop=client.loop)
                             except Exception as e:
                                 print(e)
                                 return
+                            if url_json:
+                                break
+                    if not url_json:
+                        return
                     text, user_text, media_model, media_list, quoted_status = get_twitter_status(url_json)
                     text = f'<b>Twitter Status Info</b>\n\n{text}\n\n{user_text}'
                     if len(media_model) == 0:
@@ -87,13 +93,19 @@ async def twitter_share(client: Client, message: Message):
                 else:
                     # 解析用户
                     uid = url.path.replace('/', '')
+                    url_json = None
                     with ThreadPoolExecutor() as executor:
+                        for _ in range(3):
                             try:
                                 future = client.loop.run_in_executor(executor, twitter_api.GetUser, None, uid)
                                 url_json = await asyncio.wait_for(future, timeout=30, loop=client.loop)
                             except Exception as e:
                                 print(e)
                                 return
+                            if url_json:
+                                break
+                    if not url_json:
+                        return
                     text, user_username, status_link = get_twitter_user(url_json)
                     await message.reply_photo(
                         url_json.profile_image_url_https.replace('_normal', ''),