1fb93e34c6
- 飞书消息接收与处理（文字、图片、Word 文档） - WordPress REST API 文章发布 - 图片自动上传到媒体库 - Word 文档解析与发布 - HTML 格式化与分类自动匹配 - Python CLI 工具（避免 shell 引号冲突） - Webhook 服务器（8080 端口） - 完整日志系统
276 lines
9.1 KiB
Python
276 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
WordPress 发布系统 - 分类匹配模块
|
||
根据指令或内容自动匹配 WordPress 分类
|
||
"""
|
||
|
||
import re
|
||
from modules.wp_logger import get_publish_logger, get_debug_logger
|
||
|
||
# 默认分类配置
|
||
DEFAULT_CATEGORY_ID = 7 # 随笔
|
||
DEFAULT_CATEGORY_NAME = '随笔'
|
||
|
||
# 分类关键词映射（用于 AI 自动匹配）
|
||
CATEGORY_KEYWORDS = {
|
||
12: ['ai 科普', '人工智能科普', 'ai 入门', '科普'], # Ai 科普
|
||
11: ['ai 资讯', '人工智能新闻', 'ai 新闻', '资讯', '行业动态'], # Ai 资讯
|
||
16: ['geo', '生成式引擎优化', '搜索优化'], # GEO
|
||
9: ['人工智能', 'ai', '机器学习', '深度学习', '神经网络', '大模型'], # 人工智能
|
||
5: ['技术', '教程', '开发', '编程', '代码', '技术资料'], # 技术资料
|
||
10: ['好物', '推荐', '分享', '产品', '测评'], # 好物分享
|
||
4: ['文章', '转载', '译文', '翻译'], # 文章分享
|
||
8: ['杂记', 'misc', '其他'], # 杂记
|
||
7: ['随笔', '日记', '心情', '感想'], # 随笔
|
||
1: ['关于', '网站', '联系', '声明'], # 关于我们
|
||
}
|
||
|
||
|
||
class CategoryMatcher:
|
||
"""分类匹配器"""
|
||
|
||
def __init__(self, wp_api):
|
||
"""
|
||
初始化分类匹配器
|
||
|
||
Args:
|
||
wp_api: WordPress API 客户端实例
|
||
"""
|
||
self.wp_api = wp_api
|
||
self.categories_cache = []
|
||
self.default_category_id = DEFAULT_CATEGORY_ID
|
||
|
||
self.pl = get_publish_logger()
|
||
self.dl = get_debug_logger()
|
||
|
||
def load_categories(self):
|
||
"""加载分类列表"""
|
||
self.categories_cache = self.wp_api.get_categories()
|
||
self.dl.debug(f"已加载 {len(self.categories_cache)} 个分类")
|
||
return self.categories_cache
|
||
|
||
def match_by_slug(self, slug):
|
||
"""
|
||
根据 slug 匹配分类
|
||
|
||
Args:
|
||
slug: 分类 slug
|
||
|
||
Returns:
|
||
int: 分类 ID，未找到返回默认分类
|
||
"""
|
||
if not slug:
|
||
return self.default_category_id
|
||
|
||
slug = slug.lower().strip()
|
||
|
||
for category in self.categories_cache:
|
||
if category.get('slug', '').lower() == slug:
|
||
self.dl.debug(f"通过 slug 匹配到分类：{category['name']} (ID: {category['id']})")
|
||
return category['id']
|
||
|
||
self.dl.warning(f"未找到 slug 为 '{slug}' 的分类，使用默认分类")
|
||
return self.default_category_id
|
||
|
||
def match_by_name(self, name):
|
||
"""
|
||
根据名称匹配分类
|
||
|
||
Args:
|
||
name: 分类名称
|
||
|
||
Returns:
|
||
int: 分类 ID，未找到返回默认分类
|
||
"""
|
||
if not name:
|
||
return self.default_category_id
|
||
|
||
name = name.strip()
|
||
|
||
for category in self.categories_cache:
|
||
if category.get('name', '').strip() == name:
|
||
self.dl.debug(f"通过名称匹配到分类：{category['name']} (ID: {category['id']})")
|
||
return category['id']
|
||
|
||
self.dl.warning(f"未找到名称为 '{name}' 的分类，使用默认分类")
|
||
return self.default_category_id
|
||
|
||
def match_by_instruction(self, instruction):
|
||
"""
|
||
根据指令文本匹配分类
|
||
|
||
Args:
|
||
instruction: 指令文本（如 "#分类 技术" 或 "分类：ai"）
|
||
|
||
Returns:
|
||
int: 分类 ID
|
||
"""
|
||
if not instruction:
|
||
return self.default_category_id
|
||
|
||
instruction = instruction.strip().lower()
|
||
|
||
# 匹配多种指令格式
|
||
patterns = [
|
||
r'#分类\s*([^\s#]+)', # #分类 技术
|
||
r'#category\s*([^\s#]+)', # #category tech
|
||
r'分类 [:：]\s*([^\s\n]+)', # 分类：技术
|
||
r'分类 [:：]\s*([^\s\n]+)', # 分类：tech
|
||
r'发布到\s*([^\s\n]+)', # 发布到 技术
|
||
]
|
||
|
||
for pattern in patterns:
|
||
match = re.search(pattern, instruction, re.IGNORECASE)
|
||
if match:
|
||
category_value = match.group(1).strip()
|
||
self.dl.debug(f"从指令中提取分类：{category_value}")
|
||
|
||
# 尝试匹配 slug
|
||
category_id = self.match_by_slug(category_value)
|
||
if category_id != self.default_category_id:
|
||
return category_id
|
||
|
||
# 尝试匹配名称
|
||
category_id = self.match_by_name(category_value)
|
||
if category_id != self.default_category_id:
|
||
return category_id
|
||
|
||
self.dl.warning("指令中未找到有效分类，使用默认分类")
|
||
return self.default_category_id
|
||
|
||
def match_by_content(self, title, content):
|
||
"""
|
||
根据内容自动匹配分类（AI 匹配）
|
||
|
||
Args:
|
||
title: 文章标题
|
||
content: 文章内容
|
||
|
||
Returns:
|
||
int: 分类 ID
|
||
"""
|
||
if not title and not content:
|
||
return self.default_category_id
|
||
|
||
# 合并标题和内容用于分析
|
||
text = f"{title} {content}".lower()
|
||
|
||
# 统计每个分类的关键词匹配分数
|
||
scores = {}
|
||
for cat_id, keywords in CATEGORY_KEYWORDS.items():
|
||
score = 0
|
||
for keyword in keywords:
|
||
if keyword.lower() in text:
|
||
score += 1
|
||
if score > 0:
|
||
scores[cat_id] = score
|
||
|
||
if scores:
|
||
# 返回得分最高的分类
|
||
best_cat_id = max(scores, key=scores.get)
|
||
best_score = scores[best_cat_id]
|
||
|
||
# 获取分类名称
|
||
cat_name = "未知"
|
||
for category in self.categories_cache:
|
||
if category['id'] == best_cat_id:
|
||
cat_name = category['name']
|
||
break
|
||
|
||
self.dl.debug(f"AI 匹配分类：{cat_name} (ID: {best_cat_id}, 得分：{best_score})")
|
||
return best_cat_id
|
||
|
||
self.dl.debug("内容未匹配到关键词，使用默认分类")
|
||
return self.default_category_id
|
||
|
||
def match(self, instruction=None, title=None, content=None, auto_match=True):
|
||
"""
|
||
综合匹配分类
|
||
|
||
Args:
|
||
instruction: 指令文本（优先级最高）
|
||
title: 文章标题
|
||
content: 文章内容
|
||
auto_match: 是否启用自动匹配
|
||
|
||
Returns:
|
||
int: 分类 ID
|
||
"""
|
||
# 确保分类列表已加载
|
||
if not self.categories_cache:
|
||
self.load_categories()
|
||
|
||
# 优先级 1：指令匹配
|
||
if instruction:
|
||
category_id = self.match_by_instruction(instruction)
|
||
if category_id != self.default_category_id:
|
||
self.pl.info(f"📂 分类：根据指令匹配到分类 ID {category_id}")
|
||
return category_id
|
||
|
||
# 优先级 2：自动匹配（如果启用）
|
||
if auto_match and (title or content):
|
||
category_id = self.match_by_content(title or '', content or '')
|
||
if category_id != self.default_category_id:
|
||
self.pl.info(f"📂 分类：根据内容自动匹配到分类 ID {category_id}")
|
||
return category_id
|
||
|
||
# 默认分类
|
||
self.pl.info(f"📂 分类：使用默认分类 ID {self.default_category_id}")
|
||
return self.default_category_id
|
||
|
||
def get_category_list(self):
|
||
"""
|
||
获取分类列表（用于显示）
|
||
|
||
Returns:
|
||
list: 分类信息列表
|
||
"""
|
||
if not self.categories_cache:
|
||
self.load_categories()
|
||
|
||
return [
|
||
{
|
||
'id': cat['id'],
|
||
'name': cat['name'],
|
||
'slug': cat['slug'],
|
||
'keywords': CATEGORY_KEYWORDS.get(cat['id'], [])
|
||
}
|
||
for cat in self.categories_cache
|
||
]
|
||
|
||
|
||
def create_category_matcher(wp_api):
|
||
"""创建分类匹配器实例"""
|
||
return CategoryMatcher(wp_api)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
import sys
|
||
|
||
if len(sys.argv) < 4:
|
||
print("用法：python wp_category.py <wp_url> <wp_user> <wp_password> [instruction]")
|
||
sys.exit(1)
|
||
|
||
wp_url = sys.argv[1]
|
||
wp_user = sys.argv[2]
|
||
wp_password = sys.argv[3]
|
||
instruction = sys.argv[4] if len(sys.argv) > 4 else None
|
||
|
||
from modules.wp_api import create_wp_api
|
||
api = create_wp_api(wp_url, wp_user, wp_password)
|
||
matcher = create_category_matcher(api)
|
||
|
||
# 显示所有分类
|
||
print("可用分类列表:")
|
||
for cat in matcher.get_category_list():
|
||
print(f" ID: {cat['id']}, 名称：{cat['name']}, Slug: {cat['slug']}")
|
||
if cat['keywords']:
|
||
print(f" 关键词：{', '.join(cat['keywords'])}")
|
||
|
||
# 测试匹配
|
||
if instruction:
|
||
print(f"\n指令 '{instruction}' 匹配结果:")
|
||
category_id = matcher.match(instruction=instruction)
|
||
print(f" 匹配到分类 ID: {category_id}")
|