Add RSS news ingestion workflow

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;

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.