教你用Agent Skills复现N8N自动生成+发布公众号的流程

wechat-article-publisher/├── SKILL.md(主要指令)├──.env(环境变量)└── scripts/ └── generate_image.py(图片生成脚本) └── markdown2html.py(markdown转html模板脚本) └── publish_wechat.py(公众平台上传脚本)└── references/ └── 违禁词库.md(违禁词)└── templates/ └── defaultlayout.md(默认排版) └── techlayout.md(科技风排版)

# 与“最”有关最、最佳、最具、最爱、最赚、最优、最优秀、最好、最大、最大程度、最高、最高级、最高端、最奢侈、最低、最低级、最低价、最底、最便宜、史上最低价、最流行、最受欢迎、最时尚、最符合、最舒适、最先、最先进、最先进科学、最先进加工工艺、最先享受、最后、最后一波、最新、最新技术、最新科学。# 与“一”有关第一、中国第一、全网第一、销量第一、排名第一、唯一、第一品牌、NO.1、TOP.1、独一无二、全国第一、一流、一天、仅此一次(一款)、最后一波、全国X大品牌之一。# 与“级/极”有关国家级、国家级产品、全球级、宇宙级、世界级、顶级(顶尖\尖端)、顶级工艺、顶级享受、高级、极品、极佳(绝佳\绝对)、终极、极致。# 与“首/家/国”有关首个、首选、独家、独家配方、首发、全网首发、全国首发、首家、全网首家、全国首家、xx网独家、xx网首发、首次、首款、全国销量冠军、国家级产品、国家(国家免检)、国家领导人、填补国内空白、中国驰名(驰名商标)、国际品质。# 与“品牌”有关大牌、金牌、名牌、王牌、领袖品牌、世界领先、(遥遥)领先、领导者、缔造者、创领品牌、领先上市、巨星、著名、掌门人、至尊、巅峰、奢侈、优秀、资深、领袖、之王、王者、冠军。# 与“虚假”有关史无前例、前无古人、永久、万能、祖传、特效、无敌、纯天然、100%、高档、正品、真皮、超赚、精确。# 与“权威”有关老字号、中国驰名商标、特供、专供、专家推荐、质量免检、无需国家质量检测、免抽检、国家XX领导人推荐、国家XX机关推荐、使用人民币图样(央行批准除外)。# 与“欺诈”有关点击领奖、恭喜获奖、全民免单、点击有惊喜、点击获取、点击转身、点击试穿、点击翻转、领取奖品、秒杀、抢爆、再不抢就没了、不会再便宜了、没有他就xX，错过就没机会了、万人疯抢、全民疯抢/抢购、卖/抢疯了。# 与“时间”有关几天几夜、倒计时、趁现在、就、仅限、周末、周年庆、特惠趴、购物大趴、闪购、品牌团、精品团、单品团(必须有活动日期)、随时结束、随时涨价、马上降价。# 与“迷信”有关带来好运气、增强第六感、化解小人、增加事业运、招财进宝、健康富贵、提升运气、有助事业、护身、平衡正负能量、消除精神压力、调和气压、逢凶化吉、时来运转、万事亨通、旺人、旺财、助吉避凶、转富招福。# 与“医疗”有关抗癌、防癌、癌症、医生推荐、医生认证、壮阳、补肾、丰胸、滋阴补阳、全面调整人体内分泌平衡、增强或提高免疫力、助眠、失眠、消炎、可促进新陈代谢、减少红血丝、产生优化细胞结构、修复受损肌肤、抗炎、活血、解毒、抗敏、脱敏、减肥、清热解毒、清热袪湿、治疗、除菌、杀菌、抗菌、灭菌、防菌、消毒、排毒，防敏、柔敏、舒敏、缓敏、脱敏、褪敏、改善敏感肌肤、改善过敏现象、降低肌肤敏感度、胃胀蠕动、利尿、驱寒解毒、调节内分泌、延缓更年期、补肾、祛风、生发、丘疹、脓疱、手癣、甲癣、体癣、头癣、股癣、脚癣、脚气、鹅掌癣、花斑癣、牛皮癣、传染性湿疹。

importrequestsimportosfromdotenvimportload_dotenvparent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))load_dotenv(dotenv_path=os.path.join(parent_dir,'.env'), override=True)defcall_n8n_jimeng(image_prompt:str): """ 根据图片提示词调用n8n接口生成图片，返回图片链接 :param image_prompt: 图片生成的描述提示词 :return: 图片URL链接 或 None """ n8n_webhook_url = os.getenv("N8N_WEBHOOK_URL") username = os.getenv("N8N_USERNAME") password = os.getenv("N8N_PASSWORD") payload = { "image_prompt": image_prompt } print(f"请求 payload:{payload}") response = requests.post(n8n_webhook_url, json=payload, timeout=120, auth=(username, password)) result = response.json() returnresult.get('randomImg')

# -*- coding: utf-8 -*-# 发布微信文章importrequestsimportjsonimportosfromdotenvimportload_dotenv# 加载环境变量parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))load_dotenv(dotenv_path=os.path.join(parent_dir,'.env'), override=True)# 微信 API 地址TOKEN_URL ="https://api.weixin.qq.com/cgi-bin/token"UPLOAD_URL ="https://api.weixin.qq.com/cgi-bin/material/add_material"PUBLISH_URL ="https://api.weixin.qq.com/cgi-bin/draft/add"defget_access_token(): """获取 access_token""" appid = os.getenv("WECHAT_APPID") appsecret = os.getenv("WECHAT_APPSECRET") params = { "grant_type":"client_credential", "appid": appid, "secret": appsecret } resp = requests.get(TOKEN_URL, params=params) result = resp.json() if"access_token"inresult: returnresult["access_token"] else: raiseException(f"获取 access_token 失败:{result}")defupload_image(access_token, image_path): """上传图片到临时素材""" img_response = requests.get(image_path, timeout=30) ifimg_response.status_code !=200: raiseException(f"下载图片失败:{img_response.status_code}")
files = {'media': ('image.png', img_response.content)} params = { 'access_token': access_token, 'type':'image' } resp = requests.post(UPLOAD_URL, params=params, files=files, timeout=120)
result = resp.json() if'media_id'inresult: print("✅ 上传成功！") print(json.dumps(result, indent=2, ensure_ascii=False)) returnresult['media_id'] else: raiseException(f"上传失败:{result}")defadd_draft(access_token, wechat_title, htmlContent, mediaId): """调用新版草稿箱接口 /draft/add""" articles = [{ "article_type":"news",# 文章类型，设置为"news"表示文章消息 "title": wechat_title,#文章标题 "author":"墨香异境",#文章作者信息 "content": htmlContent,#文章正文内容 "thumb_media_id": mediaId,#封面图片的媒体ID，来自上传图片素材步骤 "show_cover_pic":1,#是否显示封面图片（1=显示，0=不显示） "need_open_comment":1,#是否开启评论功能（1=开启，0=关闭） "only_fans_can_comment":0,#评论权限（1=仅粉丝，0=所有人） "auto_publish":False,#自动发布设置（false=创建草稿，true=立即发布） "publish_time":"immediate"#发布时间（"immediate"=立即发布） }] payload = {"articles": articles} headers = {"Content-Type":"application/json; charset=utf-8"} resp = requests.post( PUBLISH_URL, params={"access_token": access_token}, data=json.dumps(payload, ensure_ascii=False).encode('utf-8'), headers=headers ) result = resp.json() if'media_id'notinresult: raiseException(f"草稿上传失败:{result}") returnresult['media_id']

importmarkdownimportreimportsysdefconvert_markdown_to_html(md_content): """ 将Markdown内容转换为HTML :param md_content: Markdown格式的内容 :return: 转换后的HTML内容 """ md = markdown.Markdown(extensions=['extra','codehilite','toc']) html = md.convert(md_content) returnhtmldefextract_start_content(html_content): """ 提取Markdown内容中标题之后到第一个二级标题之前的部分 :param html_content: HTML格式的内容 :return: 提取的起始内容 """ h1_pattern =r']*>(.*?)' match_h1 = re.search(h1_pattern, html_content) # 查找第一个标签 h2_pattern =r']*>(.*?)' match_h2 = re.search(h2_pattern, html_content) ifmatch_h1andmatch_h2: end_pos = match_h1.end() start_pos = match_h2.start() # 提取标题之后到第一个二级标题之前的内容 start_content = html_content[end_pos:start_pos].strip() returnstart_content else: return""# 如果没有二级标题，返回空字符串defextract_sections(html_content): """ 从HTML内容中提取二级标题及其对应内容 :param html_content: HTML格式的内容 :return: 包含二级标题和对应内容的列表 """ # 查找所有标签及其内容 h2_pattern =r']*>(.*?)' sections = [] # 找到所有二级标题 h2_matches =list(re.finditer(h2_pattern, html_content)) fori,matchinenumerate(h2_matches): title =match.group(1) start_pos =match.end() # 找到下一个二级标题的位置 end_pos =len(html_content) ifi <len(h2_matches) -1: end_pos = h2_matches[i +1].start() # 提取标题对应的内容 content = html_content[start_pos:end_pos].strip() sections.append({ 'title': title, 'content': content }) returnsectionsdefapply_default_template(sections, start_content, template_path): """ 应用排版模板 :param sections: 包含二级标题和对应内容的列表 :param start_content: 标题之后到第一个二级标题之前的内容 :param template_path: 模板文件路径 :return: 应用模板后的HTML内容 """ withopen(template_path,'r', encoding='utf-8')asf: template = f.read() # 分离模板的各个部分 parts = template.split('# ') guide_section ="" title_part ="" content_part ="" divider_part ="" interaction_part ="" forpartinparts: ifpart.startswith("引导关注"): guide_section = part[5:].strip() elifpart.startswith("二级文章标题"): title_part = part[7:].strip() elifpart.startswith("二级标题内容"): content_part = part[7:].strip() elifpart.startswith("分割线"): divider_part = part[3:].strip() elifpart.startswith("文末互动"): interaction_part = part[4:].strip() # 构建最终HTML final_html =f"

" # 添加文章开头部分（标题之后到第一个二级标题之前的内容） final_html +=f"

" # 为每个section应用模板样式 forsectioninsections: # 替换标题模板中的占位符 title_html = title_part.replace('{{title_content}}', section['title']) # 替换内容模板中的占位符 content_html = content_part.replace('{{content}}', section['content']) final_html +=f"

" final_html +=f"

" final_html +=f"

" final_html +=f"

" returnfinal_htmldefmain(input_file, template_path, output_file): # 读取原始Markdown文件 withopen(input_file,'r', encoding='utf-8')asf: md_content = f.read() # 转换为HTML html_content = convert_markdown_to_html(md_content) start_content = extract_start_content(html_content) # 提取章节 sections = extract_sections(html_content) # 应用默认模板 final_html = apply_default_template(sections, start_content, template_path) # 保存排版后的文件 withopen(output_file,'w', encoding='utf-8')asf: f.write(final_html) print(f"排版完成！文件已保存为：{output_file}")if__name__ =="__main__": # 参数获取 iflen(sys.argv) !=4: print("使用方法: python markdown2html.py ") sys.exit(1)
input_file = sys.argv[1] template_path = sys.argv[2] output_file = sys.argv[3] main(input_file, template_path, output_file)

WECHAT_APPID=""WECHAT_APPSECRET=""N8N_WEBHOOK_URL=""N8N_USERNAME="N8N_PASSWORD=""