这是测试文章的正文内容。
" + }] + + # 执行测试消息发送（web_hook函数内部会处理字典类型） + from jobs.webhook import MessageWebHook, web_hook + # 使用类型忽略注释，因为web_hook函数会处理字典和Article对象 + test_hook = MessageWebHook( # type: ignore + task=message_task, + feed=feed, + articles=mock_articles # type: ignore + ) + + result = web_hook(test_hook, is_test=True) + + return success_response( + data={ + "task_id": task_id, + "feed_name": feed.mp_name, + "test_article": mock_articles[0], + "result": result + }, + message=f"测试消息已发送到 {feed.mp_name}" + ) + except HTTPException: + raise except Exception as e: + print_error(e) return error_response(code=500, message=str(e)) @router.get("/{task_id}/run", summary="执行单个消息任务详情") async def run_message_task( diff --git a/core/notice/__init__.py b/core/notice/__init__.py index 930618cf..ac698029 100644 --- a/core/notice/__init__.py +++ b/core/notice/__init__.py @@ -14,7 +14,7 @@ def notice( webhook_url, title, text,notice_type: str=None): - text: 消息内容 """ if len(str(webhook_url)) == 0: - print('未提供webhook_url') + raise ValueError('未提供webhook_url') return if 'qyapi.weixin.qq.com' in webhook_url: notice_type = 'wechat' @@ -35,4 +35,4 @@ def notice( webhook_url, title, text,notice_type: str=None): elif notice_type == 'custom': send_custom_message(webhook_url, title, text) else: - print('不支持的通知类型') \ No newline at end of file + raise ValueError(f'不支持的通知类型: {notice_type}') \ No newline at end of file diff --git a/jobs/mps.py b/jobs/mps.py index 08e98c4e..ba6e5dce 100644 --- a/jobs/mps.py +++ b/jobs/mps.py @@ -36,31 +36,48 @@ from core.models.message_task import MessageTask # from core.queue import TaskQueue from .webhook import web_hook interval=int(cfg.get("interval",60)) # 每隔多少秒执行一次 -def do_job(mp=None,task:MessageTask=None): +def do_job(mp=None,task:MessageTask=None,isTest=False): # TaskQueue.add_task(test,info=datetime.now().strftime("%Y-%m-%d %H:%M:%S")) # print("执行任务", task.mps_id) - print("执行任务") + print(f"执行任务 (测试模式: {isTest})") all_count=0 - wx=WxGather().Model() - try: - wx.get_Articles(mp.faker_id,CallBack=UpdateArticle,Mps_id=mp.id,Mps_title=mp.mp_name, MaxPage=1,Over_CallBack=Update_Over,interval=interval) - except Exception as e: - print_error(e) - # raise - finally: - count=wx.all_count() - all_count+=count - from jobs.webhook import MessageWebHook - tms=MessageWebHook(task=task,feed=mp,articles=wx.articles) - web_hook(tms) - print_success(f"任务({task.id})[{mp.mp_name}]执行成功,{count}成功条数") + if isTest: + # 测试模式使用模拟数据 + from datetime import datetime, timedelta + mock_articles = [{ + "id": "test-article-001", + "mp_id": mp.id, + "title": "测试文章标题", + "pic_url": "https://via.placeholder.com/300x200", + "url": "https://example.com/test-article", + "description": "这是一篇测试文章的描述内容，用于测试webhook功能是否正常。", + "publish_time": (datetime.now() - timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S"), + "content": "这是测试文章的正文内容。
" + }] + count = 1 + else: + wx=WxGather().Model() + try: + wx.get_Articles(mp.faker_id,CallBack=UpdateArticle,Mps_id=mp.id,Mps_title=mp.mp_name, MaxPage=1,Over_CallBack=Update_Over,interval=interval) + except Exception as e: + print_error(e) + # raise + finally: + count=wx.all_count() + mock_articles = wx.articles + all_count+=count + + from jobs.webhook import MessageWebHook + tms=MessageWebHook(task=task,feed=mp,articles=mock_articles) + web_hook(tms, is_test=isTest) + print_success(f"任务({task.id})[{mp.mp_name}]执行成功,{count}成功条数") from core.queue import TaskQueue def add_job(feeds:list[Feed]=None,task:MessageTask=None,isTest=False): if isTest: TaskQueue.clear_queue() for feed in feeds: - TaskQueue.add_task(do_job,feed,task) + TaskQueue.add_task(do_job,feed,task,isTest) if isTest: print(f"测试任务，{feed.mp_name}，加入队列成功") reload_job() diff --git a/jobs/webhook.py b/jobs/webhook.py index bc1d8590..235dc8d8 100644 --- a/jobs/webhook.py +++ b/jobs/webhook.py @@ -48,19 +48,24 @@ def send_message(hook: MessageWebHook) -> str: message = parser.render(data) # 这里可以添加发送消息的具体实现 print("发送消息:", message) - notice(hook.task.web_hook_url, hook.task.name, message) + try: + notice(hook.task.web_hook_url, hook.task.name, message) + except Exception as e: + logger.error(f"发送消息失败: {e}") + raise ValueError(f"发送消息失败: {e}") return message -def call_webhook(hook: MessageWebHook) -> str: +def call_webhook(hook: MessageWebHook, is_test: bool = False) -> str: """ 调用webhook接口发送数据 - + 参数: hook: MessageWebHook对象，包含任务、订阅源和文章信息 - + is_test: 是否为测试模式，测试模式下使用模拟数据 + 返回: str: 调用结果信息 - + 异常: ValueError: 当webhook调用失败时抛出 """ @@ -91,31 +96,48 @@ def call_webhook(hook: MessageWebHook) -> str: "now": "{{ now }}" } """ - + # 检查template是否需要content template_needs_content = "content" in template.lower() - + # 根据content_format处理内容 content_format = cfg.get("webhook.content_format", "html") logger.info(f'Content将以{content_format}格式发送') - processed_articles = [] - for article in hook.articles: - if isinstance(article, dict) and "content" in article and article["content"]: - processed_article = article.copy() - # 只有template需要content时才进行格式转换 - if template_needs_content: - processed_article["content"] = format_content(processed_article["content"], content_format) - processed_articles.append(processed_article) - else: - processed_articles.append(article) - + + # 测试模式下使用模拟数据 + if is_test: + from datetime import timedelta + logger.info("使用模拟数据测试webhook") + mock_article = { + "id": "test-article-001", + "mp_id": hook.feed.id if hook.feed else "test-mp-id", + "title": "测试文章标题", + "pic_url": "https://via.placeholder.com/300x200", + "url": "https://example.com/test-article", + "description": "这是一篇测试文章的描述内容，用于测试webhook功能是否正常。", + "publish_time": (datetime.now() - timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S"), + "content": "这是测试文章的正文内容。
" + } + processed_articles = [mock_article] + else: + processed_articles = [] + for article in hook.articles: + if isinstance(article, dict) and "content" in article and article["content"]: + processed_article = article.copy() + # 只有template需要content时才进行格式转换 + if template_needs_content: + processed_article["content"] = format_content(processed_article["content"], content_format) + processed_articles.append(processed_article) + else: + processed_articles.append(article) + data = { "feed": hook.feed, "articles": processed_articles, "task": hook.task, "now": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } - + # 预处理content字段 import json def process_content(content): @@ -125,7 +147,7 @@ def call_webhook(hook: MessageWebHook) -> str: json_escaped = json.dumps(content, ensure_ascii=False) # 去掉外层引号避免重复 return json_escaped[1:-1] - + # 处理articles中的content字段，进行JSON转义 if "articles" in data: for i, article in enumerate(data["articles"]): @@ -134,11 +156,11 @@ def call_webhook(hook: MessageWebHook) -> str: data["articles"][i]["content"] = process_content(article["content"]) elif hasattr(article, "content"): setattr(data["articles"][i], "content", process_content(getattr(article, "content"))) - + parser = TemplateParser(template) - + payload = parser.render(data) - # logger.info(payload) + logger.info(f'Webhook payload: {payload}') # 检查web_hook_url是否为空 if not hook.task.web_hook_url: @@ -181,16 +203,16 @@ def call_webhook(hook: MessageWebHook) -> str: except Exception as e: raise ValueError(f"Webhook调用失败: {str(e)}") -def web_hook(hook:MessageWebHook): +def web_hook(hook:MessageWebHook, is_test:bool = False): """ 根据消息类型路由到对应的处理函数 - + 参数: hook: MessageWebHook对象，包含任务、订阅源和文章信息 - + is_test: 是否为测试模式，测试模式下使用模拟数据 返回: 对应处理函数的返回结果 - + 异常: ValueError: 当消息类型未知时抛出 """ @@ -200,36 +222,58 @@ def web_hook(hook:MessageWebHook): if len(hook.articles)<=0: # raise ValueError("没有更新到文章") logger.warning("没有更新到文章") - return + return for article in hook.articles: if isinstance(article, dict): # 如果是字典类型，直接使用 + def process_field_value(field_name, article): + value = article.get(field_name, "") + if field_name == "publish_time" and value: + # 如果已经是格式化的字符串，直接返回 + if isinstance(value, str) and "-" in value and ":" in value: + return value + # 如果是时间戳整数，转换为字符串 + try: + if isinstance(value, (int, float)): + return datetime.fromtimestamp(value).strftime("%Y-%m-%d %H:%M:%S") + except (ValueError, TypeError, OSError): + pass + return value + return value + processed_article = { - field.name: ( - datetime.fromtimestamp(article[field.name]).strftime("%Y-%m-%d %H:%M:%S") - if field.name == "publish_time" and field.name in article - else article.get(field.name, "") - ) + field.name: process_field_value(field.name, article) for field in Article.__table__.columns } else: # 如果是Article对象，使用getattr获取属性 + def process_field_value_obj(field_name, article): + value = getattr(article, field.name, "") + if field_name == "publish_time" and value: + # 如果已经是格式化的字符串，直接返回 + if isinstance(value, str) and "-" in value and ":" in value: + return value + # 如果是时间戳整数，转换为字符串 + try: + if isinstance(value, (int, float)): + return datetime.fromtimestamp(value).strftime("%Y-%m-%d %H:%M:%S") + except (ValueError, TypeError, OSError): + pass + return value + return value + processed_article = { - field.name: ( - datetime.fromtimestamp(getattr(article, field.name)).strftime("%Y-%m-%d %H:%M:%S") - if field.name == "publish_time" - else getattr(article, field.name) - ) + field.name: process_field_value_obj(field.name, article) for field in Article.__table__.columns } processed_articles.append(processed_article) - + hook.articles = processed_articles - + if hook.task.message_type == 0: # 发送消息 return send_message(hook) elif hook.task.message_type == 1: # 调用webhook - return call_webhook(hook) + return call_webhook(hook, is_test) else: raise ValueError(f"未知的消息类型: {hook.task.message_type}") except Exception as e: diff --git a/web_ui/src/api/messageTask.ts b/web_ui/src/api/messageTask.ts index 2b7bd2bf..1076b8d7 100644 --- a/web_ui/src/api/messageTask.ts +++ b/web_ui/src/api/messageTask.ts @@ -16,6 +16,10 @@ export const RunMessageTask = (id: string,isTest:boolean=false) => { return http.get