Add Inoreader webhook support

Author: aturret

app/routers/inoreader.py

@@ -40,10 +40,10 @@
 
 
 async def get_inoreader_item_async(
-    data: Optional[Dict] = None,
-    trigger: bool = False,
-    params: Optional[Dict] = None,
-    # filters: Optional[Dict] = None,
+        data: Optional[Dict] = None,
+        trigger: bool = False,
+        params: Optional[Dict] = None,
+        # filters: Optional[Dict] = None,
 ) -> None:
     stream_id = None
     use_inoreader_content = True
@@ -76,19 +76,19 @@ async def get_inoreader_item_async(
 
 
 async def process_inoreader_data(
-    data: list,
-    use_inoreader_content: bool,
-    telegram_channel_id: Union[int, str] = default_telegram_channel_id,
-    stream_id: str = None,
+        data: list,
+        use_inoreader_content: bool,
+        telegram_channel_id: Union[int, str] = default_telegram_channel_id,
+        stream_id: str = None,
 ):
     for item in data:
         url_type_item = await get_url_metadata(item["aurl"])
         url_type_dict = url_type_item.to_dict()
         logger.debug(f"ino original: {use_inoreader_content}")
         if (
-            use_inoreader_content is True
-            or url_type_dict["content_type"] == "unknown"
-            # or url_type_dict["source"] == "zhihu"
+                use_inoreader_content is True
+                or url_type_dict["content_type"] == "unknown"
+                # or url_type_dict["source"] == "zhihu"
         ):
             is_video = url_type_dict["content_type"] == "video"
             content_type = url_type_dict["content_type"] if is_video else "social_media"
@@ -118,6 +118,10 @@ async def process_inoreader_data(
             )
 
 
+async def get_inoreader_webhook_data(data: dict):
+    result = data["items"]
+    return result
+
 # @router.post("/", dependencies=[Security(verify_api_key)])
 # async def inoreader_repost_webhook(request: Request, background_tasks: BackgroundTasks):
 #     data = await request.json()
@@ -142,3 +146,11 @@ async def inoreader_trigger_webhook(request: Request):
     params = request.query_params
     await get_inoreader_item_async(trigger=True, params=params)
     return "ok"
+
+
+@router.post("/webhook", dependencies=[Security(verify_api_key)])
+async def inoreader_tag_webhook(request: Request):
+    data = await request.json()
+    data = await Inoreader.process_items_data(data)
+    await process_inoreader_data(data=data, use_inoreader_content=True, telegram_channel_id=default_telegram_channel_id)
+    return "ok"

app/services/inoreader/__init__.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from urllib.parse import quote
 
 import httpx
@@ -97,7 +98,7 @@ async def get_api_item_data(
         tag: str = None,
         feed: str = None,
         params: dict = None,
-    ) -> dict:
+    ) -> Optional[dict | list]:
         stream_id = Inoreader.get_stream_id(stream_type=stream_type, tag=tag, feed=feed)
         request_url = INOREADER_CONTENT_URL + stream_id
         default_params = {
@@ -112,23 +113,29 @@ async def get_api_item_data(
         resp = await Inoreader.get_api_info(url=request_url, params=params)
         logger.debug(resp.text)
         data = resp.json()
+        data = await Inoreader.process_items_data(data)
+        return data
+
+    @staticmethod
+    async def process_items_data(data: dict) -> Optional[dict | list]:
         expression = """
-                    items[].{
-                    "aurl": canonical[0].href,
-                    "title": title,
-                    "author": origin.title,
-                    "author_url": origin.htmlUrl,
-                    "content": summary.content,
-                    "category": categories[-1],
-                    "message": comments[0].commentBody,
-                    "timestamp": updated
-                    }
-                """
+                            items[].{
+                            "aurl": canonical[0].href,
+                            "title": title,
+                            "author": origin.title,
+                            "author_url": origin.htmlUrl,
+                            "content": summary.content,
+                            "category": categories[-1],
+                            "message": comments[0].commentBody,
+                            "timestamp": updated
+                            }
+                        """
         data = jmespath.search(expression, data)
         for item in data:
             item["category"] = item["category"].split("/")[-1]
         return data
 
+
     @staticmethod
     async def get_api_info(
         url: str,