新增公众号发文 api

applications/api/gzh_api.py

@@ -0,0 +1,32 @@
+import json
+import requests
+
+from applications.utils.async_http_client import AsyncHttpClient
+
+
+app_id = "wx96cd2914fe4d0ae9"
+app_secret = "04f8d6c3a4d61adbee64e9854197b2a9"
+
+
+class GzhApi:
+    def __init__(self, app_id, app_secret):
+        self.app_id = app_id
+        self.app_secret = app_secret
+        self.access_token = None
+
+    async def get_access_token(self):
+        url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={self.app_id}&secret={self.app_secret}"
+        headers = {"Content-Type": "application/json;charset=UTF-8"}
+        async with AsyncHttpClient() as client:
+            response = await client.get(url, headers=headers)
+
+        self.access_token = response.json()["access_token"]
+        return self.access_token
+
+    async def add_draft(self, data: Dict):
+        url = f"https://api.weixin.qq.com/cgi-bin/draft/add?access_token={self.access_token}"
+        headers = {"Content-Type": "application/json;charset=UTF-8"}
+        async with AsyncHttpClient() as client:
+            response = await client.post(url, json=data, headers=headers)
+
+        return response.json()

applications/tasks/crawler_tasks/crawler_account_manager.py

@@ -115,7 +115,9 @@ class WeixinAccountManager(CrawlerAccountManager):
 
         # 计算发文频率
         publish_times = pd.to_numeric(dataframe["publish_time"], errors="coerce")
-        publish_times = publish_times.replace([float("inf"), float("-inf")], pd.NA).dropna()
+        publish_times = publish_times.replace(
+            [float("inf"), float("-inf")], pd.NA
+        ).dropna()
         if len(publish_times) >= 2:
             dates = pd.to_datetime(publish_times, unit="s").dt.normalize()
             days_delta = max(int((dates.max() - dates.min()).days) + 1, 1)