微信小程序业务域名获取

core/utils/feishu_data_async.py

@@ -118,7 +118,7 @@ class FeishuDataAsync:
         :param spreadsheet_token: 电子表格token
         :param sheet_id: 工作表ID
         :param ranges: 数据范围（如"A1:C10"）
-        :param values: 要更新的数据列表
+        :param values: 要更新的数据列表,必须是二维数组 [['祝福有福有财  ', 'https://54dc5260-2.xuweiying.cn'], ['祝福有福有财  ', 'https://54dc5260.xuweiying.cn']]
         :return: 更新结果
         """
         access_token = await self._get_access_token()
@@ -134,13 +134,11 @@ class FeishuDataAsync:
                 "valueRange":
                     {
                         "range": f"{sheet_id}!{ranges}",
-                        "values": [values]
+                        "values": values
                     },
 
             }
         async with self.session.post(url, headers=headers, json=payload,ssl=False) as response:
-            if response.status != 200:
-                error_text = await response.text()
             if response.status != 200:
                 error_text = await response.text()
                 raise Exception(f"更新表格数据失败: {error_text}")

core/utils/helpers.py

@@ -60,7 +60,7 @@ async def generate_titles(sheet_id: str,video_obj: Dict,logger,aliyun_log):
 async def insert_safe_data(sheet_id: str, values: List):
     spreadsheet_token = "U5dXsSlPOhiNNCtEfgqcm1iYnpf"
     async with FeishuDataAsync() as feishu:
-        await feishu.insert_values(spreadsheet_token=spreadsheet_token, sheet_id=sheet_id,ranges="A2:Z2",values=values)
+        await feishu.insert_values(spreadsheet_token=spreadsheet_token, sheet_id=sheet_id,ranges="A2:Z2",values=[values])
 
 async def is_near_next_day(threshold_minutes: int = 3) -> bool:
         """

scripts/office/wx_getDomainInfo.py

@@ -0,0 +1,73 @@
+import asyncio
+from readline import insert_text
+
+import requests
+import json
+
+from core.utils.feishu_data_async import FeishuDataAsync
+
+class WechatDomainFetcher:
+    """
+    微信小程序域名信息获取类
+    一次性任务
+    """
+
+    def __init__(self):
+        pass
+
+    def get_access_token(self, appid, secret):
+        """获取微信接口访问令牌"""
+        url = "https://api.weixin.qq.com/cgi-bin/token"
+        params = {
+            "grant_type": "client_credential",
+            "appid": appid,
+            "secret": secret
+        }
+
+        try:
+            response = requests.get(url, params=params)
+            print(response)
+            return response.json().get("access_token")
+        except Exception as e:
+            print(f"获取微信令牌失败: {str(e)}")
+            return None
+
+    def get_domain_info(self, appid, secret):
+        """获取小程序域名信息"""
+        access_token = self.get_access_token(appid, secret)
+        if not access_token:
+            return None
+
+        url = f"https://api.weixin.qq.com/wxa/getwxadevinfo?access_token={access_token}"
+
+        try:
+            response = requests.get(url)
+            return response.json()
+        except Exception as e:
+            print(f"获取域名信息失败: {str(e)}")
+            return None
+
+async def main():
+    async with FeishuDataAsync() as feishu_data:
+        config = await feishu_data.get_values("HJlWwCCzwis5KIk24DOc0dtjnhh", "5178f1")
+    # 初始化微信域名信息获取器
+    domain_fetcher = WechatDomainFetcher()
+
+    for _,app_name, appid, secret in config[1:]:
+        if not appid or not secret:
+            continue
+        domain_info = domain_fetcher.get_domain_info(appid, secret)
+        if not domain_info:
+            continue
+        bizdomain = domain_info.get('bizdomain')
+        if bizdomain:
+            insert_data = list(map(lambda x: [app_name, x], bizdomain))
+            print(insert_data)
+            async with FeishuDataAsync() as feishu_data:
+                    await feishu_data.insert_values("TxA2wpGZHiuLl2kMMokcaU9Mnlb", "d3a349", "A2:B", insert_data)
+        else:
+            print(f"小程序 {app_name} 无法获取域名信息")
+
+
+if __name__ == '__main__':
+    asyncio.run(main())