新增内容判断引擎文章发布接口

2026-09-11 16:55:32 +08:00
parent dd34718c8c
commit dd0326e30e
12 changed files with 860 additions and 21 deletions
@@ -0,0 +1,305 @@
+import crypto from 'crypto';
+import express from 'express';
+import { withTransaction } from './db.js';
+
+const SOURCE_SYSTEM = 'amaba-content-core';
+const SITE_PROFILE = 'human';
+const MAX_TITLE_LENGTH = 200;
+const MAX_BODY_LENGTH = 200_000;
+const MAX_SUMMARY_LENGTH = 1_000;
+const MAX_ARRAY_ITEMS = 30;
+
+class ApiError extends Error {
+  constructor(status, code, message, retryable = false) {
+    super(message);
+    this.status = status;
+    this.code = code;
+    this.retryable = retryable;
+  }
+}
+
+function fail(res, error) {
+  const status = error instanceof ApiError ? error.status : 500;
+  const code = error instanceof ApiError ? error.code : 'INTERNAL_ERROR';
+  const message = error instanceof ApiError ? error.message : '网站内部错误';
+  const retryable = error instanceof ApiError ? error.retryable : true;
+  return res.status(status).json({ success: false, error: { code, message, retryable } });
+}
+
+function text(value, max, field, required = false) {
+  const result = typeof value === 'string' ? value.trim() : '';
+  if (required && !result) throw new ApiError(400, 'VALIDATION_ERROR', `${field}不能为空`);
+  if (result.length > max) throw new ApiError(413, 'CONTENT_TOO_LARGE', `${field}超过长度限制`);
+  return result;
+}
+
+function stringArray(value, field) {
+  if (value == null) return [];
+  if (!Array.isArray(value)) throw new ApiError(400, 'VALIDATION_ERROR', `${field}必须是数组`);
+  if (value.length > MAX_ARRAY_ITEMS) throw new ApiError(400, 'VALIDATION_ERROR', `${field}最多${MAX_ARRAY_ITEMS}项`);
+  return value.map((item) => text(item, 100, `${field}元素`, true));
+}
+
+function jsonObject(value, field) {
+  if (value == null) return {};
+  if (typeof value !== 'object' || Array.isArray(value)) throw new ApiError(400, 'VALIDATION_ERROR', `${field}必须是对象`);
+  return value;
+}
+
+function limitedJson(value, field, maxLength = 50_000) {
+  const serialized = JSON.stringify(value);
+  if (serialized.length > maxLength) throw new ApiError(413, 'CONTENT_TOO_LARGE', `${field}超过长度限制`);
+  return value;
+}
+
+function optionalHttpUrl(value, field) {
+  const result = text(value, 2_000, field);
+  if (!result) return null;
+  try {
+    const parsed = new URL(result);
+    if (!['http:', 'https:'].includes(parsed.protocol)) throw new Error('protocol');
+    return parsed.toString();
+  } catch {
+    throw new ApiError(400, 'VALIDATION_ERROR', `${field}必须是有效的HTTP或HTTPS地址`);
+  }
+}
+
+function safeSlug(value, sourceTaskId) {
+  const supplied = text(value, 160, 'article.slug');
+  if (supplied) {
+    const normalized = supplied.toLowerCase();
+    if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(normalized)) {
+      throw new ApiError(400, 'VALIDATION_ERROR', 'article.slug只能包含小写字母、数字和连字符');
+    }
+    return normalized;
+  }
+  const taskPart = sourceTaskId.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-|-$/g, '').slice(0, 120);
+  return `article-${taskPart || crypto.createHash('sha256').update(sourceTaskId).digest('hex').slice(0, 24)}`;
+}
+
+export function escapeHtml(value) {
+  return String(value)
+    .replaceAll('&', '&amp;')
+    .replaceAll('<', '&lt;')
+    .replaceAll('>', '&gt;')
+    .replaceAll('"', '&quot;')
+    .replaceAll("'", '&#39;');
+}
+
+export function plainTextToParagraphHtml(value) {
+  return String(value)
+    .replace(/\r\n?/g, '\n')
+    .split('\n')
+    .map((paragraph) => paragraph.trim())
+    .filter(Boolean)
+    .map((paragraph) => `<p>${escapeHtml(paragraph)}</p>`)
+    .join('\n');
+}
+
+export function validateIngestRequest(body, idempotencyKey) {
+  const payload = jsonObject(body, '请求体');
+  const article = jsonObject(payload.article, 'article');
+  const quality = jsonObject(payload.quality, 'quality');
+  const provenance = jsonObject(payload.provenance, 'provenance');
+  const semantic = jsonObject(payload.semantic, 'semantic');
+  const sourceSystem = text(payload.source_system, 100, 'source_system', true);
+  const sourceTaskId = text(payload.source_task_id, 191, 'source_task_id', true);
+  const sourceVersionId = text(payload.source_version_id, 191, 'source_version_id', true);
+  const profile = text(payload.site_profile, 32, 'site_profile', true);
+  const key = text(idempotencyKey, 191, 'Idempotency-Key', true);
+  const expectedKey = `${sourceTaskId}:${sourceVersionId}:${SITE_PROFILE}`;
+
+  if (sourceSystem !== SOURCE_SYSTEM) throw new ApiError(403, 'SOURCE_NOT_ALLOWED', `只允许source_system=${SOURCE_SYSTEM}`);
+  if (profile !== SITE_PROFILE) throw new ApiError(403, 'SITE_PROFILE_MISMATCH', `site_profile必须为${SITE_PROFILE}`);
+  if (quality.human_confirmed !== true) throw new ApiError(403, 'HUMAN_CONFIRMATION_REQUIRED', '仅允许发布人工确认的终稿');
+  if (key !== expectedKey) throw new ApiError(400, 'VALIDATION_ERROR', `Idempotency-Key必须为${expectedKey}`);
+
+  const sourceVersionNumber = Number(payload.source_version_number);
+  if (!Number.isInteger(sourceVersionNumber) || sourceVersionNumber < 1) {
+    throw new ApiError(400, 'VALIDATION_ERROR', 'source_version_number必须是正整数');
+  }
+  const contentHash = text(payload.content_hash, 64, 'content_hash', true).toLowerCase();
+  if (!/^[a-f0-9]{64}$/.test(contentHash)) throw new ApiError(400, 'VALIDATION_ERROR', 'content_hash必须是64位SHA256十六进制值');
+  const publishMode = text(payload.publish_mode || 'publish', 16, 'publish_mode');
+  if (!['publish', 'draft'].includes(publishMode)) throw new ApiError(400, 'VALIDATION_ERROR', 'publish_mode仅支持publish或draft');
+
+  const bodyPlain = text(article.body_plain, MAX_BODY_LENGTH, 'article.body_plain', true).replace(/\r\n?/g, '\n');
+  const keywords = stringArray(article.keywords, 'article.keywords');
+  const remainingIssueLevels = stringArray(quality.remaining_issue_levels, 'quality.remaining_issue_levels');
+  const knowledgeRefs = Array.isArray(provenance.knowledge_refs) ? provenance.knowledge_refs : [];
+  if (knowledgeRefs.length > MAX_ARRAY_ITEMS) throw new ApiError(400, 'VALIDATION_ERROR', `provenance.knowledge_refs最多${MAX_ARRAY_ITEMS}项`);
+  const entities = stringArray(semantic.entities, 'semantic.entities');
+  const questions = stringArray(semantic.questions, 'semantic.questions');
+  if (semantic.facts != null && !Array.isArray(semantic.facts)) throw new ApiError(400, 'VALIDATION_ERROR', 'semantic.facts必须是数组');
+  if ((semantic.facts || []).length > MAX_ARRAY_ITEMS) throw new ApiError(400, 'VALIDATION_ERROR', `semantic.facts最多${MAX_ARRAY_ITEMS}项`);
+
+  return {
+    requestId: text(payload.request_id, 191, 'request_id', true),
+    sourceSystem,
+    sourceTaskId,
+    sourceVersionId,
+    sourceVersionNumber,
+    siteProfile: profile,
+    idempotencyKey: key,
+    contentHash,
+    publishMode,
+    title: text(article.title, MAX_TITLE_LENGTH, 'article.title', true),
+    bodyPlain,
+    contentHtml: plainTextToParagraphHtml(bodyPlain),
+    summary: text(article.summary, MAX_SUMMARY_LENGTH, 'article.summary'),
+    requestedSlug: safeSlug(article.slug, sourceTaskId),
+    primaryKeyword: text(article.primary_keyword, 255, 'article.primary_keyword'),
+    keywords,
+    contentType: text(article.content_type, 64, 'article.content_type'),
+    direction: text(article.direction, 191, 'article.direction'),
+    targetAudience: text(article.target_audience, 500, 'article.target_audience'),
+    authorName: text(article.author_name || '阿美巴', 191, 'article.author_name'),
+    coverImageUrl: optionalHttpUrl(article.cover_image_url, 'article.cover_image_url'),
+    canonicalUrl: optionalHttpUrl(article.canonical_url, 'article.canonical_url'),
+    humanConfirmed: true,
+    confirmedAt: text(quality.confirmed_at, 40, 'quality.confirmed_at', true),
+    quality: limitedJson({ ...quality, remaining_issue_levels: remainingIssueLevels }, 'quality'),
+    provenance: limitedJson({ ...provenance, knowledge_refs: knowledgeRefs }, 'provenance'),
+    semantic: limitedJson({ ...semantic, entities, questions, facts: semantic.facts || [] }, 'semantic'),
+  };
+}
+
+function publicationResult(row, publicSiteUrl, contentHash = row.content_hash, revision = row.revision) {
+  return {
+    publication_id: row.id,
+    status: row.status,
+    url: `${publicSiteUrl}/insights/${encodeURIComponent(row.slug)}`,
+    revision: Number(revision),
+    content_hash: contentHash,
+    published_at: row.publish_time || row.updated_at,
+  };
+}
+
+export function createMysqlArticleRepository() {
+  return {
+    async upsert(input, publicSiteUrl) {
+      return withTransaction(async (tx) => {
+        const priorRequest = await tx.one(`SELECT * FROM article_ingest_requests WHERE idempotency_key=? FOR UPDATE`, [input.idempotencyKey]);
+        if (priorRequest) {
+          if (priorRequest.content_hash !== input.contentHash) {
+            throw new ApiError(409, 'CONTENT_CONFLICT', '相同Idempotency-Key对应了不同正文哈希');
+          }
+          const article = await tx.one(`SELECT * FROM articles WHERE id=?`, [priorRequest.article_id]);
+          if (!article) throw new ApiError(500, 'INTERNAL_ERROR', '幂等记录对应的文章不存在', true);
+          return { created: false, replayed: true, result: publicationResult(article, publicSiteUrl, priorRequest.content_hash, priorRequest.revision) };
+        }
+
+        const existing = await tx.one(
+          `SELECT * FROM articles WHERE source_system=? AND source_task_id=? AND site_profile=? FOR UPDATE`,
+          [input.sourceSystem, input.sourceTaskId, input.siteProfile],
+        );
+        const now = new Date().toISOString();
+        const status = input.publishMode === 'publish' ? 'published' : 'draft';
+        let articleId;
+        let slug;
+        let revision;
+
+        if (existing) {
+          if (input.sourceVersionNumber <= Number(existing.source_version_number || 0)) {
+            throw new ApiError(409, 'CONTENT_CONFLICT', '新发布版本号必须高于网站当前版本');
+          }
+          articleId = existing.id;
+          slug = existing.slug;
+          revision = Number(existing.revision || 1) + 1;
+          await tx.execute(`UPDATE articles SET
+            title=?, subtitle=?, category=?, cover_url=?, content_html=?, body_plain=?, publish_time=?, status=?,
+            source_version_id=?, source_version_number=?, idempotency_key=?, content_hash=?, revision=?, primary_keyword=?,
+            keywords_json=?, content_type=?, direction=?, target_audience=?, author_name=?, human_confirmed=?, confirmed_at=?,
+            request_id=?, canonical_url=?, quality_json=?, provenance_json=?, semantic_json=?, updated_at=? WHERE id=?`, [
+            input.title, input.summary, input.direction || input.contentType || '通用深度', input.coverImageUrl,
+            input.contentHtml, input.bodyPlain, status === 'published' ? now : existing.publish_time, status,
+            input.sourceVersionId, input.sourceVersionNumber, input.idempotencyKey, input.contentHash, revision,
+            input.primaryKeyword, JSON.stringify(input.keywords), input.contentType, input.direction, input.targetAudience,
+            input.authorName, 1, input.confirmedAt, input.requestId, input.canonicalUrl, JSON.stringify(input.quality),
+            JSON.stringify(input.provenance), JSON.stringify(input.semantic), now, articleId,
+          ]);
+        } else {
+          const slugOwner = await tx.one(`SELECT id FROM articles WHERE slug=?`, [input.requestedSlug]);
+          if (slugOwner) throw new ApiError(409, 'CONTENT_CONFLICT', '请求的slug已被其他文章使用');
+          articleId = `art-${crypto.randomUUID()}`;
+          slug = input.requestedSlug;
+          revision = 1;
+          await tx.execute(`INSERT INTO articles (
+            id, slug, title, subtitle, category, cover_url, content_html, body_plain, publish_time, recommend_weight,
+            status, source_system, source_task_id, source_version_id, source_version_number, site_profile,
+            idempotency_key, content_hash, revision, primary_keyword, keywords_json, content_type, direction,
+            target_audience, author_name, human_confirmed, confirmed_at, request_id, canonical_url,
+            quality_json, provenance_json, semantic_json, created_at, updated_at
+          ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, [
+            articleId, slug, input.title, input.summary, input.direction || input.contentType || '通用深度',
+            input.coverImageUrl, input.contentHtml, input.bodyPlain, status === 'published' ? now : null, 0,
+            status, input.sourceSystem, input.sourceTaskId, input.sourceVersionId, input.sourceVersionNumber,
+            input.siteProfile, input.idempotencyKey, input.contentHash, revision, input.primaryKeyword,
+            JSON.stringify(input.keywords), input.contentType, input.direction, input.targetAudience,
+            input.authorName, 1, input.confirmedAt, input.requestId, input.canonicalUrl,
+            JSON.stringify(input.quality), JSON.stringify(input.provenance), JSON.stringify(input.semantic), now, now,
+          ]);
+        }
+
+        await tx.execute(`INSERT INTO article_ingest_requests (idempotency_key, article_id, source_version_id, content_hash, revision, created_at) VALUES (?,?,?,?,?,?)`,
+          [input.idempotencyKey, articleId, input.sourceVersionId, input.contentHash, revision, now]);
+        const row = await tx.one(`SELECT * FROM articles WHERE id=?`, [articleId]);
+        return { created: !existing, replayed: false, result: publicationResult(row, publicSiteUrl) };
+      });
+    },
+  };
+}
+
+function secureTokenEquals(actual, expected) {
+  if (!actual || !expected) return false;
+  const actualDigest = crypto.createHash('sha256').update(actual).digest();
+  const expectedDigest = crypto.createHash('sha256').update(expected).digest();
+  return crypto.timingSafeEqual(actualDigest, expectedDigest);
+}
+
+export function createContentIngestRouter({
+  repository = createMysqlArticleRepository(),
+  token = process.env.CONTENT_INGEST_TOKEN || '',
+  publicSiteUrl = process.env.PUBLIC_SITE_URL || 'https://www.ameiba.cn',
+  rateLimitMax = Number(process.env.CONTENT_INGEST_RATE_LIMIT || 30),
+} = {}) {
+  const router = express.Router();
+  const buckets = new Map();
+  const baseUrl = publicSiteUrl.replace(/\/+$/, '');
+
+  router.use((req, res, next) => {
+    if (!token) return fail(res, new ApiError(503, 'SERVICE_UNAVAILABLE', '内容发布接口未配置', true));
+    const now = Date.now();
+    const key = String(req.headers['x-forwarded-for'] || req.ip || req.socket.remoteAddress || 'unknown').split(',')[0].trim();
+    const bucket = buckets.get(key);
+    if (!bucket || now - bucket.startedAt >= 60_000) buckets.set(key, { startedAt: now, count: 1 });
+    else if (++bucket.count > rateLimitMax) return fail(res, new ApiError(429, 'RATE_LIMITED', '请求过于频繁，请稍后再试', true));
+
+    const authorization = String(req.headers.authorization || '');
+    const supplied = authorization.startsWith('Bearer ') ? authorization.slice(7).trim() : '';
+    if (!secureTokenEquals(supplied, token)) return fail(res, new ApiError(401, 'UNAUTHORIZED', '发布凭证无效'));
+    next();
+  });
+
+  router.get('/health', (req, res) => res.json({
+    status: 'ok',
+    service: 'amaba-human-website',
+    site_profile: SITE_PROFILE,
+    api_version: 'v1',
+  }));
+
+  router.post('/articles/upsert', async (req, res) => {
+    try {
+      const input = validateIngestRequest(req.body, req.headers['idempotency-key']);
+      const outcome = await repository.upsert(input, baseUrl);
+      res.status(outcome.created ? 201 : 200).json({ success: true, ...outcome.result });
+    } catch (error) {
+      if (!(error instanceof ApiError)) console.error('[ContentIngest]', error?.code || error?.message || 'unknown error');
+      fail(res, error);
+    }
+  });
+
+  return router;
+}
+
+export { ApiError, MAX_BODY_LENGTH, MAX_TITLE_LENGTH, SITE_PROFILE, SOURCE_SYSTEM };
@@ -60,6 +60,50 @@ export async function initDB() {
    const [rows] = await pool.query(`SHOW COLUMNS FROM global_config LIKE ?`, [column]);
    if (!rows.length) await pool.query(`ALTER TABLE global_config ADD COLUMN \`${column}\` LONGTEXT`);
  }
+
+  const articleColumns = {
+    source_system: 'VARCHAR(100) NULL',
+    source_task_id: 'VARCHAR(191) NULL',
+    source_version_id: 'VARCHAR(191) NULL',
+    source_version_number: 'INT NULL',
+    site_profile: 'VARCHAR(32) NULL',
+    idempotency_key: 'VARCHAR(191) NULL',
+    content_hash: 'VARCHAR(64) NULL',
+    body_plain: 'LONGTEXT NULL',
+    revision: 'INT NOT NULL DEFAULT 1',
+    primary_keyword: 'VARCHAR(255) NULL',
+    keywords_json: 'LONGTEXT NULL',
+    content_type: 'VARCHAR(64) NULL',
+    direction: 'VARCHAR(191) NULL',
+    target_audience: 'VARCHAR(500) NULL',
+    author_name: 'VARCHAR(191) NULL',
+    human_confirmed: 'TINYINT(1) NOT NULL DEFAULT 0',
+    confirmed_at: 'VARCHAR(40) NULL',
+    request_id: 'VARCHAR(191) NULL',
+    canonical_url: 'LONGTEXT NULL',
+    quality_json: 'LONGTEXT NULL',
+    provenance_json: 'LONGTEXT NULL',
+    semantic_json: 'LONGTEXT NULL',
+  };
+  for (const [column, definition] of Object.entries(articleColumns)) {
+    const [rows] = await pool.query(`SHOW COLUMNS FROM articles LIKE ?`, [column]);
+    if (!rows.length) await pool.query(`ALTER TABLE articles ADD COLUMN \`${column}\` ${definition}`);
+  }
+
+  const [idempotencyIndexes] = await pool.query(`SHOW INDEX FROM articles WHERE Key_name='uq_articles_idempotency_key'`);
+  if (!idempotencyIndexes.length) await pool.query(`ALTER TABLE articles ADD UNIQUE INDEX uq_articles_idempotency_key (idempotency_key)`);
+  const [sourceIndexes] = await pool.query(`SHOW INDEX FROM articles WHERE Key_name='uq_articles_source_task_site'`);
+  if (!sourceIndexes.length) await pool.query(`ALTER TABLE articles ADD UNIQUE INDEX uq_articles_source_task_site (source_system, source_task_id, site_profile)`);
+
+  await pool.query(`CREATE TABLE IF NOT EXISTS article_ingest_requests (
+    idempotency_key VARCHAR(191) PRIMARY KEY,
+    article_id VARCHAR(191) NOT NULL,
+    source_version_id VARCHAR(191) NOT NULL,
+    content_hash VARCHAR(64) NOT NULL,
+    revision INT NOT NULL,
+    created_at VARCHAR(40) NOT NULL,
+    INDEX idx_article_ingest_article_id (article_id)
+  ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`);
 }

 function normalizeSql(sql) { return sql.replace(/INSERT OR IGNORE/gi, 'INSERT IGNORE'); }
@@ -75,4 +119,34 @@ export function getDB() {
  };
 }

+export async function withTransaction(work) {
+  if (!pool) throw new Error('Database is not initialized');
+  const connection = await pool.getConnection();
+  try {
+    await connection.beginTransaction();
+    const tx = {
+      async one(sql, params = []) {
+        const [rows] = await connection.query(normalizeSql(sql), params);
+        return rows[0] || null;
+      },
+      async all(sql, params = []) {
+        const [rows] = await connection.query(normalizeSql(sql), params);
+        return rows;
+      },
+      async execute(sql, params = []) {
+        const [result] = await connection.query(normalizeSql(sql), params);
+        return result;
+      },
+    };
+    const result = await work(tx);
+    await connection.commit();
+    return result;
+  } catch (error) {
+    await connection.rollback();
+    throw error;
+  } finally {
+    connection.release();
+  }
+}
+
 export { UPLOAD_DIR, DB_HOST, DB_PORT, DB_NAME };
@@ -10,6 +10,7 @@ import crypto from 'crypto';
 import PDFDocument from 'pdfkit';
 import { initDB, getDB, UPLOAD_DIR } from './db.js';
 import { signToken, verifyToken, authMiddleware } from './middleware/auth.js';
+import { createContentIngestRouter } from './content-ingest.js';

 const __dirname = path.dirname(fileURLToPath(import.meta.url));
 const app = express();
@@ -124,6 +125,7 @@ function rateLimit({ windowMs = 60_000, max = 30 } = {}) {
 }
 const publicRateLimit = rateLimit({ windowMs: 60_000, max: 60 });
 app.use('/api/diagnosis', publicRateLimit);
+app.use('/api/internal/content/v1', createContentIngestRouter());

 app.post('/api/contact/leads', (req, res) => {
  const { name, company = '', phone, need = '' } = req.body || {};
@@ -556,6 +558,12 @@ app.get('/api/cms/contact-leads', (req,res)=>{

 app.use((err, req, res, next) => {
  if (res.headersSent) return next(err);
+  if (err?.type === 'entity.too.large') {
+    if (req.path.startsWith('/api/internal/content/v1/')) {
+      return res.status(413).json({ success: false, error: { code: 'CONTENT_TOO_LARGE', message: '请求体超过长度限制', retryable: false } });
+    }
+    return res.status(413).json({ code: 413, msg: '请求体超过长度限制' });
+  }
  if (err?.message?.includes('仅支持') || err?.code === 'LIMIT_FILE_SIZE' || err?.code === 'LIMIT_FILE_COUNT') {
    return res.status(400).json({ code: 400, msg: err.message || '上传文件不符合要求' });
  }
@@ -17,6 +17,8 @@ export function authMiddleware(req, res, next) {
  if (req.path.startsWith('/admin') || req.path.startsWith('/uploads') || req.path === '/health' || req.path === '/') return next();
  // 诊断相关全部公开（问卷、会话、报告、留资、分享、PDF）- 但CMS诊断后台需鉴权
  if (req.path.startsWith('/api/diagnosis/')) return next();
+  // 内容判断引擎使用独立的服务端 Bearer Token，由专用路由自行鉴权。
+  if (req.path.startsWith('/api/internal/content/v1/')) return next();
  // 所有 CMS 管理接口（包括 GET）都必须鉴权；只有下方聚合公开接口例外。
  if (req.path === '/api/auth/login' || req.path === '/api/cms/public' || req.path === '/api/contact/leads') return next();
  // 上传也需鉴权