From 6781d44631bbfcc5b1f853d7b2b4a50dfa4a2bc8 Mon Sep 17 00:00:00 2001 From: wuyanwanwu <104009119+wuyanwanwu@users.noreply.github.com.> Date: Mon, 10 Aug 2026 00:02:25 +0800 Subject: [PATCH] Add RSS news ingestion workflow --- db/migrations/004_n8n_ingestion.sql | 199 ++++++++++++++++++++++++ n8n/README.md | 23 +++ n8n/workflows/02-fetch-news-to-web.json | 108 +++++++++++++ 3 files changed, 330 insertions(+) create mode 100644 db/migrations/004_n8n_ingestion.sql create mode 100644 n8n/workflows/02-fetch-news-to-web.json diff --git a/db/migrations/004_n8n_ingestion.sql b/db/migrations/004_n8n_ingestion.sql new file mode 100644 index 0000000..84b7920 --- /dev/null +++ b/db/migrations/004_n8n_ingestion.sql @@ -0,0 +1,199 @@ +-- n8n article ingestion helper and verified starter RSS sources. +-- PostgreSQL 12+. Safe to run more than once in DBeaver. + +BEGIN; + +CREATE OR REPLACE FUNCTION public.ingest_news_article( + p_source_id BIGINT, + p_title TEXT, + p_summary TEXT, + p_url TEXT, + p_author TEXT, + p_language_code VARCHAR, + p_region VARCHAR, + p_category VARCHAR, + p_topics JSONB, + p_keywords JSONB, + p_importance_score INTEGER, + p_published_at TIMESTAMPTZ, + p_raw_payload JSONB, + p_skip BOOLEAN DEFAULT FALSE +) +RETURNS TABLE ( + report_id BIGINT, + article_id BIGINT, + report_date DATE, + item_count INTEGER +) +LANGUAGE plpgsql +AS $$ +DECLARE + v_article_id BIGINT; + v_report_id BIGINT; + v_report_date DATE := (CURRENT_TIMESTAMP AT TIME ZONE 'Asia/Shanghai')::DATE; + v_item_count INTEGER; + v_topics TEXT[]; + v_keywords TEXT[]; +BEGIN + IF COALESCE(p_skip, FALSE) THEN + RETURN; + END IF; + + IF p_source_id IS NULL OR NULLIF(BTRIM(p_title), '') IS NULL + OR NULLIF(BTRIM(p_url), '') IS NULL THEN + RETURN; + END IF; + + SELECT COALESCE(ARRAY_AGG(value), ARRAY[]::TEXT[]) + INTO v_topics + FROM JSONB_ARRAY_ELEMENTS_TEXT(COALESCE(p_topics, '[]'::JSONB)); + + SELECT COALESCE(ARRAY_AGG(value), ARRAY[]::TEXT[]) + INTO v_keywords + FROM JSONB_ARRAY_ELEMENTS_TEXT(COALESCE(p_keywords, '[]'::JSONB)); + + INSERT INTO news_articles ( + source_id, title, summary, url, author, language_code, + translation_status, region, category, topics, keywords, + importance_score, published_at, raw_payload + ) + VALUES ( + p_source_id, + BTRIM(p_title), + NULLIF(BTRIM(COALESCE(p_summary, '')), ''), + BTRIM(p_url), + NULLIF(BTRIM(COALESCE(p_author, '')), ''), + CASE WHEN p_language_code = 'zh' THEN 'zh' ELSE 'en' END, + CASE WHEN p_language_code = 'zh' THEN 'not_needed' ELSE 'skipped' END, + CASE WHEN p_region = 'domestic' THEN 'domestic' ELSE 'international' END, + CASE + WHEN p_category IN ( + 'general', 'ai_agent', 'agri_hardware', + 'agri_solutions', 'agri_services', 'policy_market' + ) THEN p_category + ELSE 'general' + END, + v_topics, + v_keywords, + LEAST(100, GREATEST(0, COALESCE(p_importance_score, 50))), + COALESCE(p_published_at, CURRENT_TIMESTAMP), + p_raw_payload + ) + ON CONFLICT (url) DO UPDATE SET + source_id = EXCLUDED.source_id, + title = EXCLUDED.title, + summary = COALESCE(EXCLUDED.summary, news_articles.summary), + author = COALESCE(EXCLUDED.author, news_articles.author), + language_code = EXCLUDED.language_code, + region = EXCLUDED.region, + category = EXCLUDED.category, + topics = EXCLUDED.topics, + keywords = EXCLUDED.keywords, + importance_score = EXCLUDED.importance_score, + published_at = EXCLUDED.published_at, + raw_payload = EXCLUDED.raw_payload, + updated_at = CURRENT_TIMESTAMP + RETURNING id INTO v_article_id; + + INSERT INTO daily_reports ( + report_date, title, introduction, status, email_status + ) + VALUES ( + v_report_date, + '每日资讯热点 - ' || TO_CHAR(v_report_date, 'YYYY-MM-DD'), + '自动抓取的国内外 AI、Agent、智慧农业硬件、落地方案与农业服务资讯。', + 'published', + 'skipped' + ) + ON CONFLICT (report_date) DO UPDATE SET + title = EXCLUDED.title, + introduction = EXCLUDED.introduction, + status = 'published', + updated_at = CURRENT_TIMESTAMP + RETURNING id INTO v_report_id; + + INSERT INTO report_articles ( + report_id, article_id, section, display_order, is_highlight + ) + SELECT + v_report_id, + v_article_id, + category, + 100 - importance_score, + importance_score >= 85 + FROM news_articles + WHERE id = v_article_id + ON CONFLICT (report_id, article_id) DO UPDATE SET + section = EXCLUDED.section, + display_order = EXCLUDED.display_order, + is_highlight = EXCLUDED.is_highlight; + + SELECT COUNT(*)::INTEGER + INTO v_item_count + FROM report_articles + WHERE report_articles.report_id = v_report_id; + + UPDATE daily_reports + SET item_count = v_item_count, + updated_at = CURRENT_TIMESTAMP + WHERE id = v_report_id; + + UPDATE news_sources + SET last_fetched_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE id = p_source_id; + + RETURN QUERY + SELECT v_report_id, v_article_id, v_report_date, v_item_count; +END; +$$; + +INSERT INTO news_sources ( + name, homepage_url, feed_url, region, category, priority, enabled +) +VALUES + ( + '量子位', + 'https://www.qbitai.com/', + 'https://www.qbitai.com/feed', + 'domestic', 'ai_agent', 90, TRUE + ), + ( + '智慧农业（中英文）', + 'https://www.smartag.net.cn/', + 'https://www.smartag.net.cn/CN/rss_dqml_2096-8094.xml', + 'domestic', 'agri_solutions', 88, TRUE + ), + ( + 'Google Blog', + 'https://blog.google/', + 'https://blog.google/rss/', + 'international', 'general', 78, TRUE + ), + ( + 'USDA Agricultural Research Service', + 'https://www.ars.usda.gov/news-events/', + 'https://www.ars.usda.gov/rss/?productName=Research%20News', + 'international', 'agri_solutions', 82, TRUE + ), + ( + 'AGCO Newsroom', + 'https://news.agcocorp.com/', + 'https://news.agcocorp.com/news?pagetemplate=rss', + 'international', 'agri_hardware', 84, TRUE + ), + ( + 'FAO Newsroom', + 'https://www.fao.org/newsroom/en', + 'https://www.fao.org/feeds/fao-newsroom-rss', + 'international', 'agri_services', 80, TRUE + ) +ON CONFLICT (feed_url) DO UPDATE SET + name = EXCLUDED.name, + homepage_url = EXCLUDED.homepage_url, + region = EXCLUDED.region, + category = EXCLUDED.category, + priority = EXCLUDED.priority, + updated_at = CURRENT_TIMESTAMP; + +COMMIT; diff --git a/n8n/README.md b/n8n/README.md index e13bc81..329c59a 100644 --- a/n8n/README.md +++ b/n8n/README.md @@ -46,3 +46,26 @@ password, and disable SSL when the server reports that it does not support SSL. The first workflow in `workflows/01-test-postgres.json` only checks this connection. It does not fetch news or send email. + +## RSS ingestion workflow + +Run `db/migrations/004_n8n_ingestion.sql` in DBeaver first. It creates the +idempotent database ingestion function and adds a verified starter set of RSS +sources. It is safe to run the whole file again. + +Then import `workflows/02-fetch-news-to-web.json` into n8n and select the same +PostgreSQL credential on both PostgreSQL nodes. Keep the workflow inactive for +the first test and click **Execute workflow**. The workflow: + +- loads every enabled row from `news_sources`; +- reads RSS feeds and keeps entries from the last 14 days; +- classifies entries with local keyword rules, without an AI API; +- upserts articles and publishes today's web report; +- does not send email. + +After the manual test succeeds and the web page shows today's report, activate +the workflow. Its schedule is 07:30 in `Asia/Shanghai`. + +New RSS sources can be added later with an `INSERT` into `news_sources`; the +workflow reads that table on every run, so the workflow itself does not need to +be edited. Set `enabled = FALSE` to pause a source without deleting it. diff --git a/n8n/workflows/02-fetch-news-to-web.json b/n8n/workflows/02-fetch-news-to-web.json new file mode 100644 index 0000000..6570eb0 --- /dev/null +++ b/n8n/workflows/02-fetch-news-to-web.json @@ -0,0 +1,108 @@ +{ + "id": "DNDfetchRss0001", + "name": "Daily News Digest - Fetch RSS to Web", + "nodes": [ + { + "parameters": {}, + "id": "manual-trigger", + "name": "手动执行", + "type": "n8n-nodes-base.manualTrigger", + "typeVersion": 1, + "position": [240, 260] + }, + { + "parameters": { + "rule": { + "interval": [ + { + "field": "cronExpression", + "expression": "0 30 7 * * *" + } + ] + } + }, + "id": "schedule-trigger", + "name": "每天 07:30", + "type": "n8n-nodes-base.scheduleTrigger", + "typeVersion": 1.2, + "position": [240, 420] + }, + { + "parameters": { + "operation": "executeQuery", + "query": "SELECT id AS source_id, name AS source_name, feed_url, region, category, priority\nFROM news_sources\nWHERE enabled = TRUE\nORDER BY priority DESC, id;", + "options": {} + }, + "id": "load-sources", + "name": "加载启用资讯源", + "type": "n8n-nodes-base.postgres", + "typeVersion": 2.5, + "position": [500, 340] + }, + { + "parameters": { + "url": "={{ $json.feed_url }}", + "options": {} + }, + "id": "read-rss", + "name": "读取 RSS", + "type": "n8n-nodes-base.rssFeedRead", + "typeVersion": 1.2, + "position": [760, 340], + "continueOnFail": true + }, + { + "parameters": { + "mode": "runOnceForEachItem", + "jsCode": "const feed = $json;\nconst source = $('加载启用资讯源').item.json;\n\nconst skipped = () => ({\n json: {\n skip: true,\n source_id: source.source_id,\n title: '',\n summary: '',\n url: '',\n author: '',\n language_code: 'en',\n region: source.region || 'international',\n category: source.category || 'general',\n topics: [],\n keywords: [],\n importance_score: Number(source.priority || 50),\n published_at: null,\n raw_payload: feed\n }\n});\n\nif (feed.error) return skipped();\n\nconst stripHtml = (value) => String(value || '')\n .replace(//gi, ' ')\n .replace(//gi, ' ')\n .replace(/<[^>]+>/g, ' ')\n .replace(/ /gi, ' ')\n .replace(/&/gi, '&')\n .replace(/</gi, '<')\n .replace(/>/gi, '>')\n .replace(/"/gi, String.fromCharCode(34))\n .replace(/'/gi, String.fromCharCode(39))\n .replace(/\\s+/g, ' ')\n .trim();\n\nconst title = stripHtml(feed.title);\nconst summary = stripHtml(feed.contentSnippet || feed.content || feed.description).slice(0, 1800);\nconst url = String(feed.link || feed.guid || '').trim();\nif (!title || !url || !/^https?:\\/\\//i.test(url)) return skipped();\n\nconst publishedRaw = feed.isoDate || feed.pubDate || feed.published || feed.updated || null;\nconst publishedDate = publishedRaw ? new Date(publishedRaw) : null;\nconst publishedAt = publishedDate && !Number.isNaN(publishedDate.getTime())\n ? publishedDate.toISOString()\n : null;\nconst maxAgeMs = 14 * 24 * 60 * 60 * 1000;\nif (publishedDate && Date.now() - publishedDate.getTime() > maxAgeMs) return skipped();\n\nconst text = `${title} ${summary}`.toLowerCase();\nconst keywordGroups = {\n ai_agent: [\n 'ai', 'artificial intelligence', 'machine learning', 'deep learning',\n 'large language model', 'llm', 'agent', 'agentic', 'generative ai',\n '人工智能', '大模型', '智能体', '机器学习', '深度学习', '生成式', '具身智能'\n ],\n agri_hardware: [\n 'agricultural machinery', 'farm machinery', 'sensor', 'drone', 'robot',\n 'tractor', 'harvester', 'iot', 'satellite', 'precision equipment',\n '传感器', '无人机', '机器人', '农机', '拖拉机', '收割机', '物联网', '卫星', '智能装备'\n ],\n agri_solutions: [\n 'smart farm', 'smart farming', 'precision agriculture', 'digital agriculture',\n 'greenhouse', 'irrigation', 'traceability', 'farm management',\n '智慧农业', '智慧农场', '精准农业', '数字农业', '温室', '灌溉', '水肥一体化', '追溯',\n '解决方案', '项目落地', '示范项目'\n ],\n agri_services: [\n 'agricultural service', 'farm service', 'agronomy', 'extension service',\n 'farm saas', 'crop insurance', 'agricultural finance', 'supply chain',\n '农业服务', '农技服务', '社会化服务', '托管服务', '农业 saas', '农业保险',\n '农业金融', '供应链', '运维服务'\n ],\n policy_market: [\n 'agricultural policy', 'farm policy', 'subsidy', 'regulation', 'investment',\n 'funding', 'tender', 'procurement', 'market outlook',\n '农业政策', '补贴', '监管', '投资', '融资', '招标', '采购', '市场行情'\n ]\n};\n\nconst escapeRegex = (value) => value.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\$&');\nconst containsKeyword = (keyword) => {\n if (/^[a-z0-9+#.-]+$/i.test(keyword) && keyword.length <= 3) {\n return new RegExp(`\\\\b${escapeRegex(keyword)}\\\\b`, 'i').test(text);\n }\n return text.includes(keyword.toLowerCase());\n};\n\nconst matchedByCategory = {};\nfor (const [category, words] of Object.entries(keywordGroups)) {\n matchedByCategory[category] = words.filter(containsKeyword);\n}\n\nconst sourceCategory = source.category || 'general';\nconst scores = Object.fromEntries(\n Object.entries(matchedByCategory).map(([category, words]) => [\n category,\n words.length + (sourceCategory === category ? 2 : 0)\n ])\n);\nlet category = sourceCategory;\nconst best = Object.entries(scores).sort((a, b) => b[1] - a[1])[0];\nif (best && best[1] > 0) category = best[0];\n\nconst matchedKeywords = [...new Set(Object.values(matchedByCategory).flat())];\nconst dedicatedSource = sourceCategory !== 'general';\nif (!dedicatedSource && matchedKeywords.length === 0) return skipped();\n\nconst matchedTopics = Object.entries(matchedByCategory)\n .filter(([, words]) => words.length > 0)\n .map(([topic]) => topic);\nconst topics = [...new Set([category, source.region, ...matchedTopics])];\nconst languageCode = /[\\u3400-\\u9fff]/.test(`${title}${summary}`) ? 'zh' : 'en';\nconst importanceScore = Math.max(0, Math.min(100,\n Number(source.priority || 50) + Math.min(12, matchedKeywords.length * 2)\n));\n\nreturn {\n json: {\n skip: false,\n source_id: source.source_id,\n source_name: source.source_name,\n title,\n summary,\n url,\n author: stripHtml(feed.creator || feed.author || ''),\n language_code: languageCode,\n region: source.region,\n category,\n topics,\n keywords: matchedKeywords,\n importance_score: importanceScore,\n published_at: publishedAt,\n raw_payload: feed\n }\n};" + }, + "id": "normalize-classify", + "name": "清洗与规则分类", + "type": "n8n-nodes-base.code", + "typeVersion": 2, + "position": [1020, 340] + }, + { + "parameters": { + "operation": "executeQuery", + "query": "SELECT *\nFROM public.ingest_news_article(\n $1::bigint, $2::text, $3::text, $4::text, $5::text,\n $6::varchar, $7::varchar, $8::varchar, $9::jsonb, $10::jsonb,\n $11::integer, $12::timestamptz, $13::jsonb, $14::boolean\n);", + "options": { + "queryReplacement": "={{ [\n $json.source_id,\n $json.title,\n $json.summary,\n $json.url,\n $json.author,\n $json.language_code,\n $json.region,\n $json.category,\n JSON.stringify($json.topics || []),\n JSON.stringify($json.keywords || []),\n $json.importance_score,\n $json.published_at,\n JSON.stringify($json.raw_payload || {}),\n $json.skip === true\n] }}" + } + }, + "id": "save-article", + "name": "保存文章并发布日报", + "type": "n8n-nodes-base.postgres", + "typeVersion": 2.5, + "position": [1280, 340] + } + ], + "connections": { + "手动执行": { + "main": [[{"node": "加载启用资讯源", "type": "main", "index": 0}]] + }, + "每天 07:30": { + "main": [[{"node": "加载启用资讯源", "type": "main", "index": 0}]] + }, + "加载启用资讯源": { + "main": [[{"node": "读取 RSS", "type": "main", "index": 0}]] + }, + "读取 RSS": { + "main": [[{"node": "清洗与规则分类", "type": "main", "index": 0}]] + }, + "清洗与规则分类": { + "main": [[{"node": "保存文章并发布日报", "type": "main", "index": 0}]] + } + }, + "active": false, + "settings": { + "timezone": "Asia/Shanghai", + "executionOrder": "v1" + }, + "versionId": "daily-news-digest-fetch-rss-v1", + "meta": { + "templateCredsSetupCompleted": false + }, + "pinData": {}, + "tags": [] +}