♻️ refactor: Modularize logic

separate code logic into multiple modules
qiu121 · Oct 28, 2024 · 5c08bf4 · 5c08bf4
1 parent ef79324
commit 5c08bf4
Show file tree

Hide file tree

Showing 5 changed files with 120 additions and 114 deletions.
diff --git a/bing_wallpaper.py b/bing_wallpaper.py
@@ -0,0 +1,18 @@
+import requests
+import json
+from typing import Optional, Tuple
+from config import BING_API, BING_URL
+
+def get_bing_wallpaper() -> Tuple[Optional[str], Optional[str], Optional[str]]:
+    response = requests.get(BING_API)
+    data = response.json()
+    print(json.dumps(data, ensure_ascii=False, indent=2))  # 打印标准 JSON 格式
+
+    if data and 'images' in data:
+        image_info = data['images'][0]
+        image_url = BING_URL + image_info['url']
+        image_description = image_info.get('copyright', 'Bing Daily Wallpaper')
+        image_title = image_info.get('title', '🔖今日必应壁纸')
+        return image_url, image_description, image_title
+    print('No wallpaper found')
+    return None, None, None
diff --git a/config.py b/config.py
@@ -0,0 +1,10 @@
+import os
+
+# 环境变量配置
+FEISHU_APP_ID = os.getenv('FEISHU_APP_ID')
+FEISHU_APP_SECRET = os.getenv('FEISHU_APP_SECRET')
+FEISHU_WEBHOOK_URL = os.getenv('FEISHU_WEBHOOK_URL')
+FEISHU_SIGNING_KEY = os.getenv('FEISHU_SIGNING_KEY')
+
+BING_URL = 'https://bing.com'
+BING_API = f'{BING_URL}/HPImageArchive.aspx?format=js&idx=0&n=1&mkt=zh-CN'
diff --git a/feushu_api.py b/feushu_api.py
@@ -0,0 +1,71 @@
+import requests
+import json
+from requests_toolbelt.multipart.encoder import MultipartEncoder
+from typing import Optional
+from config import FEISHU_APP_ID, FEISHU_APP_SECRET, FEISHU_WEBHOOK_URL, FEISHU_SIGNING_KEY
+from utils import calculate_signature
+import time
+
+def get_feishu_token() -> str:
+    token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
+    headers = {'Content-Type': 'application/json'}
+    data = json.dumps({
+        "app_id": FEISHU_APP_ID,
+        "app_secret": FEISHU_APP_SECRET
+    })
+    response = requests.post(token_url, headers=headers, data=data)
+    result = response.json()
+    return result.get('tenant_access_token', '')
+
+
+def upload_image(image_url: str, access_token: str) -> Optional[str]:
+    url = "https://open.feishu.cn/open-apis/im/v1/images"
+    image_data = requests.get(image_url).content
+    form = {'image_type': 'message', 'image': ('image.jpg', image_data)}
+    multi_form = MultipartEncoder(fields=form)
+    headers = {
+        'Authorization': f'Bearer {access_token}',
+        'Content-Type': multi_form.content_type
+    }
+    response = requests.post(url, headers=headers, data=multi_form)
+    image_key = response.json().get('data', {}).get('image_key')
+    return image_key
+
+
+def send_to_feishu(image_url: str, image_description: str, image_title: str, access_token: str) -> dict:
+    # 获取当前时间戳
+    timestamp = str(int(time.time()))
+    # 计算签名
+    sign = calculate_signature(timestamp, FEISHU_SIGNING_KEY)
+
+    headers = {'Content-Type': 'application/json'}
+    image_key = upload_image(image_url, access_token)
+    data = {
+        "timestamp": f'{timestamp}',
+        "sign": f'{sign}',
+        "msg_type": "post",
+        "content": {
+            "post": {
+                "zh_cn": {
+                    "title": f'🔖{image_title}',
+                    "content": [
+                        [
+                            {
+                                "tag": "img",
+                                "image_key": image_key
+                            },
+                            {
+                                "tag": "text",
+                                "text": f"📝{image_description}"
+                            },
+                        ]
+                    ]
+                }
+            }
+        }
+    }
+    response = requests.post(FEISHU_WEBHOOK_URL, headers=headers, data=json.dumps(data))
+    print(f'Request URL: {response.request.url}')
+    print(f'Request Method: {response.request.method}')
+    print(json.dumps(response.json(), ensure_ascii=False, indent=2))  # 打印标准 JSON 格式
+    return response.json()
diff --git a/main.py b/main.py
@@ -1,119 +1,13 @@
-import requests
-import json
-import os
-import time
-import hashlib
-import base64
-import hmac
 from datetime import datetime, timezone
-from requests_toolbelt.multipart.encoder import MultipartEncoder
-from typing import Optional, Tuple
-
-# 环境变量获取
-FEISHU_APP_ID = os.getenv('FEISHU_APP_ID')
-FEISHU_APP_SECRET = os.getenv('FEISHU_APP_SECRET')
-FEISHU_WEBHOOK_URL = os.getenv('FEISHU_WEBHOOK_URL')
-FEISHU_SIGNING_KEY = os.getenv('FEISHU_SIGNING_KEY')  # 签名密钥
-
-BING_URL = 'https://bing.com'
-BING_API = f'{BING_URL}/HPImageArchive.aspx?format=js&idx=0&n=1&mkt=zh-CN'
-
-# 获取当前时间，并设置为本地时区
-current_time = datetime.now(timezone.utc).astimezone()
-# 格式化为 年-月-日 时:分:秒 时区格式
-print(current_time.strftime('%Y-%m-%d %H:%M:%S %Z %z'))
-
-def get_bing_wallpaper() -> Tuple[Optional[str], Optional[str], Optional[str]]:
-    response = requests.get(BING_API)
-    data = response.json()
-    print(json.dumps(data, ensure_ascii=False, indent=2))  # 打印标准 JSON 格式
-
-    if data and 'images' in data:
-        image_info = data['images'][0]
-        image_url = BING_URL + image_info['url']
-        image_description = image_info.get('copyright', 'Bing Daily Wallpaper')
-        image_title = image_info.get('title', '🔖今日必应壁纸')
-        return image_url, image_description, image_title
-    print('No wallpaper found')
-    return None, None, None
-
-
-def upload_image(image_url: str, access_token: str) -> Optional[str]:
-    url = "https://open.feishu.cn/open-apis/im/v1/images"
-    image_data = requests.get(image_url).content
-    form = {'image_type': 'message', 'image': ('image.jpg', image_data)}
-    multi_form = MultipartEncoder(fields=form)
-    headers = {
-        'Authorization': f'Bearer {access_token}',
-        'Content-Type': multi_form.content_type
-    }
-    response = requests.post(url, headers=headers, data=multi_form)
-    image_key = response.json().get('data', {}).get('image_key')
-    return image_key
-
-
-def calculate_signature(timestamp: str, secret: str) -> str:
-    # 计算签名字符串：timestamp + "\n" + 密钥
-    to_sign = f'{timestamp}\n{secret}'
-    hmac_code = hmac.new(to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()
-    # 对结果进行base64处理
-    return base64.b64encode(hmac_code).decode('utf-8')
-
-
-def send_to_feishu(image_url: str, image_description: str, image_title: str, access_token: str) -> dict:
-    # 获取当前时间戳
-    timestamp = str(int(time.time()))
-    # 计算签名
-    sign = calculate_signature(timestamp, FEISHU_SIGNING_KEY)
-
-    headers = {'Content-Type': 'application/json'}
-    image_key = upload_image(image_url, access_token)
-    data = {
-        "timestamp": f'{timestamp}',
-        "sign": f'{sign}',
-        "msg_type": "post",
-        "content": {
-            "post": {
-                "zh_cn": {
-                    "title": f'🔖{image_title}',
-                    "content": [
-                        [
-                            {
-                                "tag": "img",
-                                "image_key": image_key
-                            },
-                            {
-                                "tag": "text",
-                                "text": f"📝{image_description}"
-                            },
-                        ]
-                    ]
-                }
-            }
-        }
-    }
-    response = requests.post(FEISHU_WEBHOOK_URL, headers=headers, data=json.dumps(data))
-    full_url = response.request.url
-    print(full_url)
-    print(f'请求方法: {response.request.method}')
-
-    print(json.dumps(response.json(), ensure_ascii=False, indent=2))  # 打印标准 JSON 格式
-    return response.json()
-
-
-def get_feishu_token() -> str:
-    token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
-    headers = {'Content-Type': 'application/json'}
-    data = json.dumps({
-        "app_id": FEISHU_APP_ID,
-        "app_secret": FEISHU_APP_SECRET
-    })
-    response = requests.post(token_url, headers=headers, data=data)
-    result = response.json()
-    return result.get('tenant_access_token', '')
-
+from feushu_api import get_feishu_token,send_to_feishu
+from bing_wallpaper import get_bing_wallpaper
 
 def main():
+    # 获取当前时间，并设置为本地时区
+    current_time = datetime.now(timezone.utc).astimezone()
+    # 格式化为 年-月-日 时:分:秒 时区格式
+    print(current_time.strftime('%Y-%m-%d %H:%M:%S %Z %z'))
+
     wallpaper_url, wallpaper_description, wallpaper_title = get_bing_wallpaper()
     if wallpaper_url:
         access_token = get_feishu_token()
@@ -122,6 +16,5 @@ def main():
         print(f"🔗 {wallpaper_url}")
         send_to_feishu(wallpaper_url, wallpaper_description, wallpaper_title, access_token)
 
-
 if __name__ == "__main__":
     main()
diff --git a/utils.py b/utils.py
@@ -0,0 +1,14 @@
+import base64
+import hashlib
+import hmac
+from typing import Optional
+
+
+def calculate_signature(timestamp: str, secret: Optional[str]) -> str:
+    # 计算签名字符串：timestamp + "\n" + 密钥
+    if not secret:
+        raise ValueError("Missing signing key for Feishu.")
+    to_sign = f'{timestamp}\n{secret}'
+    hmac_code = hmac.new(to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()
+    # 对结果进行base64处理
+    return base64.b64encode(hmac_code).decode('utf-8')