@clawhub-jiubanszd-53381bef3c
用于输出首帧素材、banner素材、追番tab图、国创web图、OTT图等资源位图片。适用于用户提供截图、人物图、logo后，需要先检查素材与尺寸规范，再给出裁切与设计方案并执行的场景。触发语句通常类似“输出首帧素材”“输出banner素材”“做追番tab图”“做国创web banner”“做OTT素材”“给这张...
--- name: banner-material-output description: 用于输出首帧素材、banner素材、追番tab图、国创web图、OTT图等资源位图片。适用于用户提供截图、人物图、logo后，需要先检查素材与尺寸规范，再给出裁切与设计方案并执行的场景。触发语句通常类似“输出首帧素材”“输出banner素材”“做追番tab图”“做国创web banner”“做OTT素材”“给这张截图裁图加设计”。必须保留人物脸部特征，不允许修脸或修改五官；优先保证人物胸像完整，不能卡头，必要时自动补背景；多人场景或背景复杂时可对背景做轻量弱化，并严格遵守安全区与资源位尺寸规则。 --- # 资源位裁图与设计 ## 概述 当需求是基于截图、人物素材和 logo 输出首帧图或 banner 资源位图时，使用这个 skill。 先检查素材和资源位规格，再给出裁切与设计处理计划，最后执行，并严格保护人物脸部特征与安全区。 ## 工作流 ### 1. 识别资源位和交付内容 先从需求里识别目标资源位。 再到 [references/规格说明.md](references/规格说明.md) 中匹配对应尺寸、是否需要 logo、是否必须加出品标。 如果资源位不明确，先追问再执行。 如果用户一次提多个资源位，先列清楚每个资源位对应的输出尺寸和标识要求。 ### 2. 先审素材，再动图 先检查当前是否提供了： - 截图、人物图或主视觉 - logo（如果有） - 安全区规范图（如果有） - 文案或必须出现的标识（如果有） 如果缺少 logo 或安全区规范图，不阻塞执行。 直接使用 [references/规格说明.md](references/规格说明.md) 里的默认规则继续，并在计划里说明本次采用的是默认规则。 ### 3. 先给处理计划，再执行 开始处理前，用 3 个部分说明计划： - 裁切构图：主体是谁，如何保证胸像完整，如何避免卡头 - 背景处理：背景是否保留、补全、压暗、弱化或轻微虚化 - 品牌与安全区：logo、出品标、关键信息准备放在哪里 计划要短，但要可执行。 即使用户要求直接做，也要先做一次快速检查，并用一小段话说清楚处理方案，再开始执行。 ## 构图规则 ### 优先保人物胸像完整 主角优先使用胸像或半身构图。 不要把画面做成只有大头的特写，除非用户明确要求。 不要从头顶裁断人物。 如果原图太紧，继续裁会卡头，就先补背景，不接受硬裁头部的方案。 ### 多人场景要保主次 如果画面里有多个人，先判断主角是谁。 不要让次要人物的脸抢走主视觉。 如果多人导致画面太挤，不要优先裁坏人物，先通过整理背景来缓解画面压力。 ## 人脸保护规则 这部分是硬约束。 ### 禁止修改人物脸部特征 不要修改以下内容： - 五官特征 - 脸型 - 表情 - 眼鼻口位置 - 年龄感 - 妆容风格 - 任何影响人物辨识度的脸部细节 ### 只允许轻量整体润色 只允许做这些轻量处理： - 整体光影调整 - 曝光、对比度、色彩平衡微调 - 阴影和高光的轻微整理 - 非结构性的轻微皮肤质感优化 不允许使用会造成修脸感的处理。 禁止瘦脸、大眼、改轮廓、换五官、重塑脸部结构。 如果某种工具可能影响脸部结构，就避开脸部，或给脸部加保护区域。 ## 背景处理规则 ### 卡头风险优先通过补背景解决 如果原图过紧，导致头部或肩线会被裁掉，就先补背景，把构图空间找回来。 补背景要自然，优先做环境延展，而不是让人一眼看出是硬生生补出来的。 脸部本身不能被改动。 ### 背景复杂时做轻量整理 如果背景太满、信息太杂、多人过多或画面压力太强，可以做这些轻量处理： - 压低背景对比度 - 轻微软化或模糊背景 - 降低背景亮度或饱和度 - 用轻微明暗塑形把主体从背景里拉出来 不要过度风格化背景。 不要让背景比人物、logo 或关键信息更抢眼。 ## 安全区与标识规则 如果用户没有给项目专属安全区图，就使用 [references/规格说明.md](references/规格说明.md) 里的默认安全区规则。 所有核心信息都必须放在安全区内。 人物脸部、标题、logo、出品标都要避开裁切高风险区域。 如果 logo 和出品标都要上，尽量分开摆放，避免相互打架，也不要压到脸部。 ## 输出方式 执行时，默认输出这些内容： - 识别到的资源位和最终尺寸 - 已有素材、缺失素材，以及本次采用的默认假设 - 裁切与构图计划 - 背景处理计划 - 标识与安全区处理计划 - 最终执行说明或编辑指令 ## 什么时候需要追问 只有在这些情况才追问： - 资源位不明确 - 给了多个人物，但主角不明确 - 用户要求加文案，但没有提供文案内容 ## 常见触发说法 以下说法都应该触发这个 skill： - 输出首帧素材 - 输出banner素材 - 做追番tab图 - 做国创web banner - 做OTT素材 - 给这张截图裁图加设计 - 用人物图和logo出资源位图 ## 参考文件 资源位尺寸、出品标要求和默认安全区规则见 [references/规格说明.md](references/规格说明.md)。 FILE:references/规格说明.md # 资源位规格说明 ## 通用规则 ### 默认工作流 1. 识别资源位和所需输出尺寸。 2. 检查现有素材。 3. 给出裁切与设计处理方案。 4. 方案明确后再执行。 ### 默认输入假设 如果缺少 logo，不阻塞执行。 先按该资源位的默认品牌处理规则继续，并在说明中标注 logo 待补。 如果缺少项目专属安全区规范图，不阻塞执行。 直接使用本文档中的默认安全区原则继续。 ## 人物与人脸规则 ### 保护人物脸部辨识度 不要修改以下内容： - 五官特征 - 脸型 - 表情 - 眼、鼻、口的位置 - 年龄感 - 妆容风格 - 任何影响人物辨识度的脸部细节 只允许做轻量整体优化，例如曝光、对比度、明暗平衡，以及非结构性的轻微皮肤质感整理。 ### 优先保证胸像构图 主角优先使用胸像或半身构图。 不要做成过度贴脸的大头图，除非用户明确要求。 不要接受卡头方案；如果原图太紧，应优先补背景，而不是硬裁头部。 ### 多人场景谨慎处理 先识别主角。 次要人物只能起辅助作用，不能抢主视觉。 如果背景拥挤、多人过满或画面压力过强，优先整理背景，不要优先破坏人物构图。 ## 背景处理规则 ### 构图过紧时优先补背景 如果原图会裁到头部或肩线，就先扩展或补齐背景，把可用构图空间找回来。 补背景应尽量自然，避免明显的生成痕迹。 人物脸部本身不能被改动。 ### 背景复杂时做轻量弱化 如果背景太满、太杂或信息压力过强，只做轻量处理： - 压低背景对比度 - 轻微软化或模糊背景 - 降低背景亮度或饱和度 - 用轻微明暗塑形把主体从背景中拉出来 不要使用过重的特效。 不要让背景比人物、logo 或关键信息更抢眼。 ## 安全区原则 当用户没有提供专属安全区图时，使用以下默认规则。 ### 核心安全区规则 人物脸部、标题、logo、出品标和其他核心信息，必须尽量放在中央安全区域内。 顶部、底部和左右边缘都视为裁切高风险区域，不要放关键信息。 边缘区域只适合放可牺牲的背景内容。 ### 2700×900 默认安全区规则 根据你提供的参考图，默认按以下原则执行： - 核心内容放在中间大安全区。 - 顶部和底部排除带内不要放文字、人脸和重要信息。 - 左右两侧排除区不要放核心信息。 - 右下角预留角标或品牌标识区域，不要让主体或文案压进去。 如果其他资源位没有给专属安全区图，也按同样思路等比套用： - 中间区域放核心内容 - 四周边缘只留背景 - 标识避开人物脸部和主要文案 ## 资源位规格 ### 追番 tab 主图尺寸：`1350×1123` logo 单独交付尺寸：`720×309` 主图是否带 logo：`否` 补充说明： - 主背景图和 logo 图分开输出。 - 构图需要适配移动端 tab 展示方式。 ### 国创 web 尺寸：`2700×900` logo 规则：如用户提供 logo，则按需求放置；如未提供，则先按干净构图继续，并说明 logo 缺失。 补充说明： - 如果用户给了 2700×900 的安全区示意图，优先按示意图执行。 - 如果没给，就使用本文档中的默认安全区原则。 ### 首帧 输出尺寸： - `880×495` - `1600×900` 标识要求：两个尺寸都必须加 `出品标` 补充说明： - 同一张主视觉需要兼顾两个比例下的可读性。 - 出品标不能压住脸部，也不要侵入主要标题安全区。 ### OTT 尺寸：`1920×1080` logo 规则：如无额外说明，按用户提供素材决定是否放 logo。 补充说明： - 优先保证电视端观看时的中心区域可读性。 - 核心信息尽量集中，避免边缘信息被设备裁切或遮挡。 ## 方案输出模板 执行前优先按以下结构整理方案： - 资源位与输出尺寸：`...` - 现有素材：`...` - 缺失素材与默认假设：`...` - 裁切与构图方案：`...` - 背景处理方案：`...` - 标识与安全区方案：`...` ## 什么时候必须追问 只有在以下情况才追问： - 资源位不明确 - 给了多个人物，但主角不明确 - 用户要求加文案，但没有提供文案内容
Turn a vague product, webpage, app, workflow, automation, plugin, or skill idea into an actionable kickoff plan. Use when Codex needs to scope a new project...
--- name: project-kickoff description: Turn a vague product, webpage, app, workflow, automation, plugin, or skill idea into an actionable kickoff plan. Use when Codex needs to scope a new project from scratch, define the MVP, choose a stack, break work into phases, identify risks, or convert a rough idea into concrete build steps. --- # Project Kickoff ## Overview Use this skill to turn an early idea into a buildable plan without forcing the user to already know product, design, or engineering language. Default to momentum. Make reasonable assumptions, state them, and only stop to ask questions when the answer changes architecture, cost, or delivery risk in a non-obvious way. ## Core Workflow Follow this order unless the user explicitly asks for only one part. 1. Classify the project. 2. Extract the real goal. 3. Define the smallest valuable version. 4. Choose the delivery shape. 5. Break implementation into phases. 6. Flag risks, unknowns, and decisions. 7. End with the clearest next action. ## 1. Classify The Project Map the request into one primary type before planning: - `web-page`: landing page, interactive site, dashboard, mini tool, portfolio - `web-app`: authenticated app, CRUD product, SaaS, internal tool - `skill`: reusable Codex skill with workflows, references, or scripts - `plugin`: Codex plugin or integration surface - `automation`: recurring task, scheduled workflow, inbox automation - `script-tool`: CLI helper, local utility, data processor - `content-system`: CMS-backed site, documentation hub, blog, knowledge base If the request spans multiple types, choose the build target that delivers the first usable value and mention the other types as later phases. ## 2. Extract The Real Goal Rewrite the request as a plain-language product goal: - Who is it for? - What frustrating job should it solve? - What should feel better after using it? - What makes it interesting enough to deserve building? If the user gives only a theme, infer a likely user and use case. Mark that inference explicitly. ## 3. Define The MVP Reduce the project to the smallest version that still feels complete. For the MVP: - Include only the core loop - Prefer one polished flow over many partial features - Remove admin panels, auth, analytics, AI, payments, collaboration, and syncing unless they are central to the first value - Keep the MVP demonstrable in one sitting Use this framing: - `must-have`: required for the first usable version - `nice-to-have`: useful but safely postponed - `later`: good future expansions ## 4. Choose The Delivery Shape Recommend a concrete implementation shape, not a vague tech list. Cover: - platform: static page, SPA, full-stack app, local script, skill folder, plugin scaffold - stack: HTML/CSS/JS, React/Vite, Next.js, Python, Node.js, etc. - data model: local state, JSON file, localStorage, SQLite, remote DB, or none - deployment shape: local only, static hosting, server deployment, Codex skill install Bias toward the lowest-complexity stack that supports the MVP cleanly. Examples: - Single interactive webpage -> static HTML/CSS/JS unless complexity clearly needs a framework - Small internal dashboard -> React + local mock data first - Reusable Codex planning helper -> skill with `SKILL.md` and optional references - Scheduled reporting assistant -> automation before full web app ## 5. Break Work Into Phases Always produce a phased build plan with concrete outputs. Use this default structure: ### Phase 0 Clarify the concept, constraints, examples, and success criteria. ### Phase 1 Build the MVP shell and the core interaction loop. ### Phase 2 Polish UX, persistence, error states, and useful quality improvements. ### Phase 3 Ship, validate, and add optional stretch features. For each phase, include: - objective - deliverables - implementation notes - done-when check ## 6. Flag Risks And Decisions Separate risk from uncertainty. Use short bullets for: - technical risks: stack mismatch, data complexity, auth, performance, API dependency - product risks: too broad MVP, unclear audience, novelty without utility - decision points: where one choice affects architecture or long-term cost When possible, pair each risk with a mitigation. ## 7. End With The Next Move Always end with the most helpful next action for the current moment. Choose one: - write the project scaffold - generate page structure and component list - draft the skill files - propose the database schema - create the implementation plan - start building immediately If the user seems ready to build, do not stop at planning. Transition into execution. ## Output Format Use this structure unless the user asks for something shorter: ### Project Summary - one short paragraph ### Users And Goal - target user - user problem - expected outcome ### MVP - must-have - nice-to-have - later ### Recommended Stack - platform - tech choices - persistence or data model - deployment shape ### Build Plan - phase 0 - phase 1 - phase 2 - phase 3 ### Risks And Decisions - key risk bullets ### Next Step - one clear recommended action ## Project-Type Guidance Read [references/project-patterns.md](references/project-patterns.md) when you need project-specific planning patterns, stack bias, or a sharper MVP cut for one of these types: - webpage or microsite - web app or dashboard - Codex skill - automation - script or CLI tool ## Handling Ambiguity When details are missing: - infer the simplest sensible audience - infer the smallest useful scope - prefer boring technology over ambitious architecture - surface assumptions in one short block Ask follow-up questions only when the answer changes: - hosting or deployment model - whether user accounts are required - whether external APIs are mandatory - whether the work is a local utility or a public product - whether persistence must survive across devices ## Quality Bar The plan should be specific enough that another strong builder could start implementation immediately. Avoid: - generic advice with no delivery shape - giant feature dumps - “it depends” without a recommendation - recommending frameworks because they are popular rather than necessary - overcomplicating a one-weekend project Prefer: - concrete scope cuts - named phases - opinionated but reversible recommendations - short assumptions - one decisive next move FILE:references/project-patterns.md # Project Patterns Use this reference when the request is still fuzzy and you need a stronger recommendation for scope, stack, or sequencing. ## 1. Webpage Or Microsite Best for: - interactive showcase - landing page - small utility with one core flow - portfolio or single-topic site Default bias: - start with static `HTML + CSS + JavaScript` - use `localStorage` for lightweight persistence - avoid frameworks unless state or routing complexity is real MVP checklist: - one clear page goal - one primary interaction - one polished visual direction - mobile layout - simple persistence if useful Common overbuild trap: - turning a playful single-page idea into a full app too early ## 2. Web App Or Dashboard Best for: - repeat use over time - multi-screen workflows - data entry, filtering, or tracking - authenticated or role-based systems Default bias: - React or Next.js only if multiple views or substantial state are required - mock local data before real backend wiring - add auth only if it changes the real user value MVP checklist: - one primary user role - one core workflow - one source of truth for data - clear empty state and error state Common overbuild trap: - shipping auth, settings, billing, and notifications before the main job works ## 3. Codex Skill Best for: - repeatable planning or execution workflows - domain guidance Codex should follow consistently - lightweight procedural knowledge with optional references or scripts Default bias: - keep `SKILL.md` lean and procedural - use `references/` for deeper guidance - add scripts only when deterministic execution matters MVP checklist: - clear trigger description - one reusable workflow - one strong output format - minimal but useful reference material Common overbuild trap: - writing a human README instead of an agent-facing skill ## 4. Automation Best for: - recurring summaries - scheduled checks - inbox-style monitoring - repetitive status gathering Default bias: - define trigger, cadence, workspace, and output first - keep prompts self-sufficient - avoid multi-step workflows that need frequent human rescue MVP checklist: - one repeated job - one schedule - one output destination - clear skip conditions Common overbuild trap: - trying to automate judgment-heavy work with vague prompts ## 5. Script Or CLI Tool Best for: - local file transforms - repetitive developer tasks - structured data cleanup - deterministic utility work Default bias: - prefer a simple script with clear inputs and outputs - optimize for repeatability before polish - write usage examples into the plan MVP checklist: - explicit input source - explicit output destination - predictable flags or parameters - basic error handling Common overbuild trap: - wrapping a small script in an unnecessary UI ## Decision Heuristics Choose the simpler option when these are true: - one user role - one main workflow - no cross-device sync required - no mandatory collaboration - no complex permissions - no backend-only dependency Choose the heavier option only when the MVP truly needs it: - authenticated data - multi-user collaboration - server-side integrations - long-lived storage beyond the local machine - complex search, filtering, or data relationships ## Scope-Cutting Prompts Use one or more of these internally when the user idea is too broad: - What is the smallest version someone would still finish using? - What can be removed without breaking the core loop? - What belongs in version two even if it sounds exciting now? - What can be mocked locally before building real infrastructure? - Which feature adds the most complexity per unit of user value?
Fault-tolerant multi-agent system for collaborative task execution with policy drafting, review, dispatch, parallel ministry actions, and comprehensive fault...
# Tang Agents - Multi-Agent Collaboration System
## Overview
A fault-tolerant multi-agent collaboration framework inspired by the Tang Dynasty's Three Departments and Six Ministries system. This system implements separation of powers, parallel execution, and comprehensive fault tolerance.
## Architecture
```
User Request
↓
[Zhongshu] Policy Drafting
↓
[Menxia] Review & Veto (Checks & Balances)
↓
[Shangshu] Task Dispatch
↓
[Six Ministries] Parallel Execution
- Libu: Permissions
- Hubu: Resources
- Libu-Content: Content Generation
- Bingbu: Security
- Xingbu: Compliance
- Gongbu: Deployment
↓
Output / Human Approval
```
## Installation
```bash
pip install -r requirements.txt
```
## Quick Start
```python
import asyncio
from tang_agents import TangSystem
async def main():
system = TangSystem()
# Process a task
edict_id = await system.process("Your task description")
# Check status
status = system.dashboard.get_status(edict_id)
print(status)
# Display dashboard
system.dashboard.display()
asyncio.run(main())
```
## Fault Tolerance Features
1. **Timeout Handling**: Each agent has independent timeout (3-15s)
2. **Circuit Breaker**: Auto-pause after 3 consecutive failures
3. **Graceful Degradation**: Return template data on failure
4. **Failure Isolation**: One ministry failure doesn't crash others
5. **Retry Mechanism**: Menxia review allows up to 3 retries
## License
MIT
FILE:example.py
import asyncio
from tang_agents import TangSystem
async def main():
"""Example usage of Tang Agents"""
# Initialize the system
system = TangSystem()
# Submit a task
print("=" * 60)
print("Tang Dynasty Multi-Agent System Demo")
print("=" * 60)
edict_id = await system.process("Create marketing strategy for new product launch")
# Check status
status = system.dashboard.get_status(edict_id)
print(f"\nFinal status: {status['status']}")
# Show dashboard
system.dashboard.display()
# If pending approval, you can approve it
if status['status'] == 'pending_approval':
system.approve(edict_id)
if __name__ == "__main__":
asyncio.run(main())
FILE:README.md
# Tang Dynasty Multi-Agent System
A fault-tolerant multi-agent collaboration framework inspired by the Tang Dynasty's Three Departments and Six Ministries system.
## Features
- **Three Departments (三省)**: Decision-making hub with separation of powers
- Zhongshu (中书): Policy drafting
- Menxia (门下): Review and veto power
- **Six Ministries (六部)**: Parallel execution layer
- Libu (吏部): Personnel/Permissions
- Hubu (户部): Resources/Budget
- Libu-content (礼部): Content generation
- Bingbu (兵部): Security
- Xingbu (刑部): Compliance
- Gongbu (工部): Engineering/Deployment
- **Fault Tolerance**: Circuit breaker, timeout handling, graceful degradation
- **Real-time Dashboard**: Monitor all agents and tasks
## Installation
```bash
pip install asyncio aiohttp
```
## Quick Start
```python
from tang_agents import TangSystem
import asyncio
async def main():
system = TangSystem()
# Submit a task
edict_id = await system.process("Launch marketing campaign for new anime")
# Check status
status = system.dashboard.get_status(edict_id)
print(status)
# Display dashboard
system.dashboard.display()
asyncio.run(main())
```
## Architecture
```
User Request (Edict)
↓
[ZhongshuAgent] Draft policy
↓
[MenxiaAgent] Review (can veto/reject)
↓
[ShangshuAgent] Dispatch tasks
↓
[Six Ministries] Parallel execution
↓
[ShangshuAgent] Aggregate results
↓
Output / Human approval
```
## Fault Tolerance
Each agent has built-in fault tolerance:
- **Timeout handling**: Auto-degrade if execution exceeds limit
- **Circuit breaker**: Pause agent after consecutive failures
- **Fallback strategies**: Return default/template data on failure
- **Failure isolation**: One ministry failure doesn't crash others
## License
MIT
FILE:requirements.txt
asyncio>=3.4.3
aiohttp>=3.8.0
FILE:tang_agents.py
"""
Tang Dynasty Multi-Agent System
A fault-tolerant multi-agent collaboration framework
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any, Callable
from enum import Enum
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
from datetime import datetime
from abc import ABC, abstractmethod
class Status(Enum):
PENDING = "pending"
DRAFTING = "drafting"
REVIEWING = "reviewing"
REJECTED = "rejected"
EXECUTING = "executing"
APPROVING = "pending_approval"
COMPLETED = "completed"
FAILED = "failed"
TIMEOUT = "timeout"
DEGRADED = "degraded"
@dataclass
class AgentHealth:
"""Health status monitoring for each agent"""
agent_name: str
status: str = "healthy"
last_success: datetime = field(default_factory=datetime.now)
failure_count: int = 0
total_calls: int = 0
avg_response_time: float = 0.0
@dataclass
class Edict:
"""A task/document to be processed"""
id: str
content: str
status: Status = Status.PENDING
draft: Dict = field(default_factory=dict)
review_result: Dict = field(default_factory=dict)
execution_results: Dict = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.now)
history: List[Dict] = field(default_factory=list)
retry_count: int = 0
max_retry: int = 3
timeout: int = 10
def log(self, agent: str, action: str):
self.history.append({
"agent": agent,
"action": action,
"time": datetime.now().isoformat()
})
class CircuitBreaker:
"""Circuit breaker pattern to prevent cascade failures"""
def __init__(self, failure_threshold=3, recovery_timeout=60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failures = {}
self.last_failure_time = {}
self.is_open = {}
def call(self, agent_name: str, func: Callable, *args, **kwargs):
"""Execute with circuit breaker protection"""
if self.is_open.get(agent_name, False):
if time.time() - self.last_failure_time.get(agent_name, 0) > self.recovery_timeout:
self.is_open[agent_name] = False
self.failures[agent_name] = 0
else:
raise Exception(f"Agent {agent_name} circuit open - service unavailable")
try:
result = func(*args, **kwargs)
self.failures[agent_name] = 0
return result
except Exception as e:
self.failures[agent_name] = self.failures.get(agent_name, 0) + 1
self.last_failure_time[agent_name] = time.time()
if self.failures[agent_name] >= self.failure_threshold:
self.is_open[agent_name] = True
print(f"[!] Circuit breaker opened for {agent_name}")
raise e
class BaseAgent(ABC):
"""Base agent class with fault tolerance"""
def __init__(self, name: str, timeout: int = 10):
self.name = name
self.timeout = timeout
self.health = AgentHealth(agent_name=name)
self.circuit_breaker = CircuitBreaker()
self.fallback_enabled = True
def execute_with_fallback(self, task: Dict, edict: Edict) -> Dict:
"""Execute with timeout and fallback handling"""
start_time = time.time()
try:
result = self.circuit_breaker.call(
self.name,
self._execute_with_timeout,
task,
edict
)
self._update_health(True, time.time() - start_time)
return result
except FutureTimeoutError:
print(f"[TIMEOUT] [{self.name}] Execution timeout")
self._update_health(False, time.time() - start_time)
return self._timeout_fallback(task, edict)
except Exception as e:
print(f"[ERROR] [{self.name}] Execution error: {e}")
self._update_health(False, time.time() - start_time)
return self._error_fallback(task, edict, str(e))
def _execute_with_timeout(self, task: Dict, edict: Edict) -> Dict:
"""Execute with timeout control"""
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(self.execute, task, edict)
return future.result(timeout=self.timeout)
@abstractmethod
def execute(self, task: Dict, edict: Edict) -> Dict:
"""Implement specific logic in subclasses"""
pass
def _timeout_fallback(self, task: Dict, edict: Edict) -> Dict:
"""Fallback when timeout occurs"""
print(f"[FALLBACK] [{self.name}] Using timeout fallback")
return {
"dept": self.name,
"status": "degraded",
"reason": "timeout",
"fallback_data": self._get_fallback_data(task)
}
def _error_fallback(self, task: Dict, edict: Edict, error: str) -> Dict:
"""Fallback when error occurs"""
print(f"[FALLBACK] [{self.name}] Using error fallback")
return {
"dept": self.name,
"status": "failed",
"reason": error,
"fallback_data": self._get_fallback_data(task)
}
def _get_fallback_data(self, task: Dict) -> Any:
"""Override in subclass to provide fallback data"""
return None
def _update_health(self, success: bool, response_time: float):
"""Update health status"""
self.health.total_calls += 1
if success:
self.health.last_success = datetime.now()
self.health.avg_response_time = (
(self.health.avg_response_time * (self.health.total_calls - 1) + response_time)
/ self.health.total_calls
)
else:
self.health.failure_count += 1
self.health.status = "degraded" if self.health.failure_count < 3 else "failed"
# ==================== Three Departments (三省) ====================
class ZhongshuAgent(BaseAgent):
"""Zhongshu Department: Policy drafting and decision-making"""
def __init__(self):
super().__init__("Zhongshu", timeout=15)
self.templates = {
"default": {
"strategy": "Standard approach",
"options": ["conservative", "balanced", "aggressive"],
"selected": "balanced"
}
}
def execute(self, task: Dict, edict: Edict) -> Dict:
"""Draft policy proposals"""
print(f"[Zhongshu] Drafting edict #{edict.id}...")
time.sleep(0.5)
content = task.get("content", "")
draft = {
"strategy": f"Solution for: {content}",
"options": [
{"type": "conservative", "confidence": 0.8, "cost": 100, "risk": "low"},
{"type": "balanced", "confidence": 0.9, "cost": 200, "risk": "medium"},
{"type": "aggressive", "confidence": 0.7, "cost": 500, "risk": "high"}
],
"selected": "balanced",
"estimated_time": "2 days"
}
return {"status": "success", "draft": draft}
def _get_fallback_data(self, task: Dict) -> Dict:
return self.templates["default"]
class MenxiaAgent(BaseAgent):
"""Menxia Department: Review and veto power (checks and balances)"""
def __init__(self):
super().__init__("Menxia", timeout=10)
self.rejection_count = {}
self.max_retry = 3
def execute(self, task: Dict, edict: Edict) -> Dict:
"""Review draft with power to reject (veto)"""
print(f"[Menxia] Reviewing edict #{edict.id}...")
draft = task.get("draft", {})
risks = []
if draft.get("selected") == "aggressive":
risks.append("High-risk strategy requires additional approval")
selected_option = next(
(o for o in draft.get("options", []) if o.get("type") == draft.get("selected")),
{}
)
if selected_option.get("cost", 0) > 1000:
risks.append("Budget exceeds limit - requires special approval")
# Check sensitive keywords in strategy text only (not in field names)
strategy_text = draft.get("strategy", "")
if any(word in strategy_text.lower() for word in ["sensitive", "violation", "risk"]):
risks.append("Content contains sensitive keywords")
return {
"passed": len(risks) == 0,
"risks": risks,
"seal": "Menxia_Approved" if len(risks) == 0 else None,
"review_time": datetime.now().isoformat()
}
def _error_fallback(self, task: Dict, edict: Edict, error: str) -> Dict:
"""When review fails, pass with manual review flag"""
print(f"[WARN] [Menxia] Review system error - flagging for manual review")
return {
"passed": True,
"risks": ["Review system error - manual review required"],
"seal": "Temporary_Approval",
"requires_manual_review": True
}
class ShangshuAgent(BaseAgent):
"""Shangshu Department: Coordination and dispatch"""
def __init__(self):
super().__init__("Shangshu", timeout=5)
def execute(self, task: Dict, edict: Edict) -> Dict:
"""Dispatch tasks to six ministries"""
print(f"[Shangshu] Dispatching edict #{edict.id}...")
content = task.get("content", "")
tasks = {
"libu": {"action": "permission_check", "target": content},
"hubu": {"action": "resource_allocation", "budget": 200},
"libu_content": {"action": "content_generation", "topic": content},
"bingbu": {"action": "security_scan", "level": "standard"},
"xingbu": {"action": "compliance_review", "strict": True},
"gongbu": {"action": "deployment", "execute": True}
}
return {"status": "success", "tasks": tasks}
def aggregate_results(self, edict: Edict, results: Dict) -> Dict:
"""Aggregate results from six ministries"""
print(f"[Shangshu] Aggregating results for #{edict.id}...")
failures = []
degraded = []
for dept, result in results.items():
if result.get("status") == "failed":
failures.append(dept)
elif result.get("status") == "degraded":
degraded.append(dept)
draft = edict.draft
high_risk = (
draft.get("selected") == "aggressive" or
len(failures) > 0 or
len([r for r in results.values() if r.get("requires_manual_review")]) > 0
)
if failures and len(failures) >= 3:
return {
"status": "failed",
"message": f"{len(failures)} ministries failed execution",
"failures": failures,
"degraded": degraded
}
if high_risk:
return {
"status": "pending_approval",
"message": "High risk or anomalies detected - requires approval",
"failures": failures,
"degraded": degraded,
"results": results
}
return {
"status": "completed",
"failures": failures,
"degraded": degraded,
"results": results
}
# ==================== Six Ministries (六部) ====================
class LibuAgent(BaseAgent):
"""Libu: Personnel and permissions"""
def __init__(self):
super().__init__("Libu", timeout=3)
def execute(self, task: Dict, edict: Edict) -> Dict:
print(f"[Libu] Checking permissions...")
time.sleep(0.3)
return {"dept": "Libu", "status": "success", "permission": "granted", "user_level": "LV5"}
class HubuAgent(BaseAgent):
"""Hubu: Resources and budget"""
def __init__(self):
super().__init__("Hubu", timeout=3)
def execute(self, task: Dict, edict: Edict) -> Dict:
print(f"[Hubu] Allocating resources...")
budget = task.get("budget", 0)
if budget > 1000:
raise Exception("Insufficient budget")
time.sleep(0.3)
return {"dept": "Hubu", "status": "success", "allocated": budget, "remaining": 1000-budget}
class LibuContentAgent(BaseAgent):
"""Libu (Content): Content generation"""
def __init__(self):
super().__init__("Libu-Content", timeout=8)
self.fallback_content = "[System Generated] Standard promotional copy"
def execute(self, task: Dict, edict: Edict) -> Dict:
print(f"[Libu-Content] Generating content...")
topic = task.get("topic", "")
time.sleep(0.5)
content = f"Promotional copy for '{topic}': ..."
return {"dept": "Libu-Content", "status": "success", "content": content, "word_count": 500}
def _get_fallback_data(self, task: Dict) -> str:
return self.fallback_content
class BingbuAgent(BaseAgent):
"""Bingbu: Security"""
def __init__(self):
super().__init__("Bingbu", timeout=5)
def execute(self, task: Dict, edict: Edict) -> Dict:
print(f"[Bingbu] Security scanning...")
time.sleep(0.4)
return {"dept": "Bingbu", "status": "success", "threats": [], "security_level": "A"}
class XingbuAgent(BaseAgent):
"""Xingbu: Compliance"""
def __init__(self):
super().__init__("Xingbu", timeout=5)
def execute(self, task: Dict, edict: Edict) -> Dict:
print(f"[Xingbu] Compliance review...")
time.sleep(0.4)
return {"dept": "Xingbu", "status": "success", "violations": [], "compliance_score": 98}
class GongbuAgent(BaseAgent):
"""Gongbu: Engineering and deployment"""
def __init__(self):
super().__init__("Gongbu", timeout=6)
def execute(self, task: Dict, edict: Edict) -> Dict:
print(f"[Gongbu] Deploying...")
time.sleep(0.5)
return {"dept": "Gongbu", "status": "success", "deployed": True, "api_calls": 5}
# ==================== Dashboard and System ====================
class TangDashboard:
"""Government dashboard for monitoring"""
def __init__(self):
self.edicts: Dict[str, Edict] = {}
self.agents_health: Dict[str, AgentHealth] = {}
self.agents_status = {
"Zhongshu": "idle",
"Menxia": "idle",
"Shangshu": "idle",
"Libu": "idle",
"Hubu": "idle",
"Libu-Content": "idle",
"Bingbu": "idle",
"Xingbu": "idle",
"Gongbu": "idle"
}
def submit_edict(self, content: str) -> str:
"""Submit a new edict (task)"""
edict_id = f"EDICT_{len(self.edicts) + 1:04d}"
edict = Edict(id=edict_id, content=content)
self.edicts[edict_id] = edict
print(f"\n{'='*60}")
print(f"[NEW EDICT] #{edict_id}")
print(f" Content: {content}")
print(f"{'='*60}")
return edict_id
def get_status(self, edict_id: str) -> Dict:
"""Query edict status"""
edict = self.edicts.get(edict_id)
if not edict:
return {"error": "Edict not found"}
return {
"id": edict.id,
"status": edict.status.value,
"current_step": edict.history[-1] if edict.history else "pending",
"retry_count": edict.retry_count,
"draft": edict.draft,
"review": edict.review_result,
"execution": edict.execution_results
}
def display(self):
"""Display dashboard"""
print(f"\n{'='*70}")
print("TANG GOVERNMENT DASHBOARD")
print(f"{'='*70}")
status_count = {}
for e in self.edicts.values():
status_count[e.status.value] = status_count.get(e.status.value, 0) + 1
print(f"\n[STATISTICS] {len(self.edicts)} total edicts")
for status, count in status_count.items():
print(f" {status}: {count}")
pending = [e for e in self.edicts.values() if e.status in [Status.PENDING, Status.DRAFTING]]
if pending:
print(f"\n[PENDING] ({len(pending)}):")
for e in pending[:5]:
print(f" #{e.id}: {e.content[:20]}... [{e.status.value}]")
rejected = [e for e in self.edicts.values() if e.status == Status.REJECTED]
if rejected:
print(f"\n[REJECTED] ({len(rejected)}):")
for e in rejected:
print(f" #{e.id}: {e.review_result.get('risks', [])}")
approving = [e for e in self.edicts.values() if e.status == Status.APPROVING]
if approving:
print(f"\n[PENDING APPROVAL] ({len(approving)}):")
for e in approving:
print(f" #{e.id}")
print(f"\n[AGENT STATUS]")
for agent, status in self.agents_status.items():
print(f" {agent}: {status}")
print(f"{'='*70}\n")
class TangSystem:
"""Main entry point for Tang Dynasty Multi-Agent System"""
def __init__(self):
self.zhongshu = ZhongshuAgent()
self.menxia = MenxiaAgent()
self.shangshu = ShangshuAgent()
self.ministries = {
"libu": LibuAgent(),
"hubu": HubuAgent(),
"libu_content": LibuContentAgent(),
"bingbu": BingbuAgent(),
"xingbu": XingbuAgent(),
"gongbu": GongbuAgent()
}
self.dashboard = TangDashboard()
async def process(self, content: str) -> str:
"""Process an edict through the complete workflow"""
edict_id = self.dashboard.submit_edict(content)
edict = self.dashboard.edicts[edict_id]
try:
# Stage 1: Zhongshu drafting
print(f"\n[STAGE 1] Zhongshu Drafting")
self.dashboard.agents_status["Zhongshu"] = "drafting"
draft_result = self.zhongshu.execute_with_fallback(
{"content": content},
edict
)
if draft_result.get("status") == "degraded":
edict.status = Status.DEGRADED
print(f"[WARN] Using degraded solution")
edict.draft = draft_result.get("draft", draft_result.get("fallback_data", {}))
edict.status = Status.DRAFTING
edict.log("Zhongshu", "Draft completed")
self.dashboard.agents_status["Zhongshu"] = "idle"
# Stage 2: Menxia review
print(f"\n[STAGE 2] Menxia Review")
self.dashboard.agents_status["Menxia"] = "reviewing"
while edict.retry_count < edict.max_retry:
review_result = self.menxia.execute_with_fallback(
{"draft": edict.draft},
edict
)
edict.review_result = review_result
if review_result.get("passed"):
edict.status = Status.EXECUTING
edict.log("Menxia", "Review passed")
print(f"[OK] Menxia review passed")
break
else:
edict.retry_count += 1
if edict.retry_count >= edict.max_retry:
edict.status = Status.FAILED
edict.log("Menxia", f"Rejected {edict.retry_count} times - abandoned")
print(f"[FAIL] Multiple rejections - task abandoned")
return edict_id
edict.status = Status.REJECTED
edict.log("Menxia", f"Rejected - revision #{edict.retry_count}")
print(f"[RETRY] Rejected, redrafting... ({edict.retry_count}/{edict.max_retry})")
draft_result = self.zhongshu.execute_with_fallback(
{"content": content, "revision": True},
edict
)
edict.draft = draft_result.get("draft", {})
self.dashboard.agents_status["Menxia"] = "idle"
# Stage 3: Shangshu dispatch
print(f"\n[STAGE 3] Shangshu Dispatch")
self.dashboard.agents_status["Shangshu"] = "dispatching"
dispatch_result = self.shangshu.execute_with_fallback(
{"content": content},
edict
)
tasks = dispatch_result.get("tasks", {})
edict.log("Shangshu", "Tasks dispatched")
# Stage 4: Six ministries parallel execution
print(f"\n[STAGE 4] Six Ministries Execution")
self.dashboard.agents_status["Shangshu"] = "coordinating"
results = {}
async def execute_isolated(agent_key, agent, task):
try:
self.dashboard.agents_status[agent.name] = "executing"
result = agent.execute_with_fallback(task, edict)
self.dashboard.agents_status[agent.name] = "idle"
return agent_key, result
except Exception as e:
print(f"[CRASH] [{agent.name}] {e}")
self.dashboard.agents_status[agent.name] = "failed"
return agent_key, {
"dept": agent.name,
"status": "failed",
"reason": str(e)
}
coroutines = [
execute_isolated(key, agent, tasks[key])
for key, agent in self.ministries.items()
]
completed = await asyncio.gather(*coroutines, return_exceptions=True)
for item in completed:
if isinstance(item, Exception):
print(f"[EXCEPTION] Agent execution: {item}")
continue
key, result = item
results[key] = result
edict.execution_results = results
# Stage 5: Shangshu aggregation
print(f"\n[STAGE 5] Results Aggregation")
final_result = self.shangshu.aggregate_results(edict, results)
self.dashboard.agents_status["Shangshu"] = "idle"
if final_result["status"] == "failed":
edict.status = Status.FAILED
print(f"[FAIL] #{edict_id} Execution failed: {final_result.get('message')}")
elif final_result["status"] == "pending_approval":
edict.status = Status.APPROVING
print(f"[APPROVAL] #{edict_id} Awaiting approval...")
print(f" Reason: {final_result.get('message')}")
else:
edict.status = Status.COMPLETED
print(f"\n[OK] #{edict_id} Execution completed")
if final_result.get("failures"):
print(f" [WARN] Partial failures: {final_result['failures']}")
if final_result.get("degraded"):
print(f" [INFO] Degraded: {final_result['degraded']}")
return edict_id
except Exception as e:
edict.status = Status.FAILED
print(f"\n[SYSTEM ERROR] #{edict_id}: {e}")
return edict_id
def approve(self, edict_id: str) -> bool:
"""Emperor's approval for high-risk edicts"""
edict = self.dashboard.edicts.get(edict_id)
if not edict or edict.status != Status.APPROVING:
return False
edict.status = Status.COMPLETED
edict.log("Emperor", "Approved")
print(f"\n[APPROVED] #{edict_id} Approved by emperor")
return True
# Example usage
async def main():
"""Demo run"""
system = TangSystem()
print("\n" + "="*70)
print("TEST CASE 1: Normal Task")
print("="*70)
await system.process("Launch marketing campaign for Demon Slayer anime")
print("\n" + "="*70)
print("TEST CASE 2: High-Risk Task")
print("="*70)
await system.process("Aggressive promotion of sensitive content")
print("\n" + "="*70)
print("FINAL DASHBOARD")
print("="*70)
system.dashboard.display()
if __name__ == "__main__":
asyncio.run(main())
FILE:__init__.py
# Tang Agents - Multi-Agent Collaboration System
# Import this to use the Tang Dynasty multi-agent framework
from .tang_agents import (
TangSystem,
TangDashboard,
BaseAgent,
Edict,
Status,
AgentHealth,
CircuitBreaker,
ZhongshuAgent,
MenxiaAgent,
ShangshuAgent,
LibuAgent,
HubuAgent,
LibuContentAgent,
BingbuAgent,
XingbuAgent,
GongbuAgent
)
__version__ = "1.0.0"
__author__ = "Shenzedong"
__all__ = [
"TangSystem",
"TangDashboard",
"BaseAgent",
"Edict",
"Status",
"AgentHealth",
"CircuitBreaker",
"ZhongshuAgent",
"MenxiaAgent",
"ShangshuAgent",
"LibuAgent",
"HubuAgent",
"LibuContentAgent",
"BingbuAgent",
"XingbuAgent",
"GongbuAgent"
]自动在小红书创作者最新作品下发布引导评论，促进关注或查看私信，支持网页版，操作风险低且可批量执行。
# 小红书创作者评论引流 Skill
## 名称
xiaohongshu-creator-outreach
## 描述
自动在创作者最新作品下发表评论，引导对方关注或查看私信。网页版完全支持，风险低。
## 触发方式
### 命令触发
```
/小红书评论引流
/评论创作者
/xhs-comment
```
### 自然语言
- "帮我给创作者评论"
- "开始小红书评论引流"
## 技术说明
**为什么用评论而不是私信？**
| 功能 | 网页版支持 | 风险 | 说明 |
|------|-----------|------|------|
| **私信** | ❌ 不支持 | 高 | 只能在APP端操作，且风控严格 |
| **评论** | ✅ 完全支持 | 低 | 网页版功能完整，可批量操作 |
**评论引流策略：**
1. 在创作者最新作品下留下评论
2. 评论内容引导对方查看私信或关注
3. 创作者收到评论提醒，更可能查看私信
4. 零封号风险，可安全批量操作
## 工作流程
```
读取Excel → 获取创作者最新作品 → 访问作品页 →
查找评论框 → 输入评论 → 发送 → 记录状态
```
## 输入
自动读取 `xiaohongshu-creator-finder/output/` 目录下最新的 `creators_*.xlsx` 文件。
## 输出
### 文件位置
```
output/comment_YYYYMMDD_HHMMSS.xlsx
output/comment_progress.json（进度文件）
```
### 数据字段
| 字段 | 说明 |
|------|------|
| 创作者ID | 用户唯一ID |
| 创作者名称 | 用户昵称 |
| 主页链接 | 用户主页URL |
| 作品链接 | 评论的作品URL |
| 评论内容 | 发送的评论 |
| 评论状态 | 待评论/已评论/评论失败/已跳过 |
| 评论时间 | 评论时间 |
| 备注 | 备注信息 |
## 配置说明
编辑 `config/settings.json`：
```json
{
"comment_template": "内容很棒！对你的AI作品很感兴趣，已发私信～",
"daily_limit": 10,
"min_delay": 3,
"max_delay": 8
}
```
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `comment_template` | 评论内容 | 默认引流语 |
| `daily_limit` | 每日评论上限 | 10 |
| `min_delay` | 最小延迟（秒） | 3 |
| `max_delay` | 最大延迟（秒） | 8 |
## 注意事项
1. **评论频率** - 建议10-20条/天，避免频繁操作
2. **内容合规** - 评论内容避免敏感词，保持友好
3. **Cookie复用** - 自动复用 creator-finder 的登录状态
4. **效果追踪** - 评论后需人工查看对方是否回复或关注
## 评论模板建议
**温和版：**
```
内容很棒！对你的AI作品很感兴趣，已发私信～
```
**直接版：**
```
Hi，看到你的AI作品很有意思，想聊个合作，方便看看私信吗？
```
**专业版：**
```
作品质量很高！我们在做AI短番创作者扶持，有兴趣可以了解下～
```
## 依赖
- Python 3.8+
- Playwright
- pandas
- openpyxl
```bash
pip install playwright pandas openpyxl
playwright install chromium
```
FILE:run_comment.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书创作者评论引流脚本
"""
import pandas as pd
import json
import time
import random
import os
import sys
from datetime import datetime
from playwright.sync_api import sync_playwright
# 配置
COOKIE_FILE = ".xiaohongshu_cookies.json"
INPUT_EXCEL = "xiaohongshu_ai_creators.xlsx"
OUTPUT_DIR = "output"
COMMENT_TEMPLATE = "内容很棒！对你的AI作品很感兴趣，已发私信～"
MIN_DELAY = 3
MAX_DELAY = 8
CHROME_PATH = "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe"
def load_cookies():
"""加载Cookie"""
try:
with open(COOKIE_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"加载Cookie失败: {e}")
return None
def load_creators():
"""读取创作者列表"""
try:
df = pd.read_excel(INPUT_EXCEL)
print(f"成功读取 {len(df)} 位创作者信息")
return df
except Exception as e:
print(f"读取Excel失败: {e}")
return None
def get_latest_note_link(page, user_link):
"""获取用户最新作品链接"""
try:
print(f"访问用户主页: {user_link}")
page.goto(user_link, wait_until="networkidle", timeout=30000)
time.sleep(3)
# 查找第一个笔记链接
# 小红书笔记链接格式: /explore/xxxx 或 /discovery/item/xxxx
note_links = page.locator('a[href*="/explore/"], a[href*="/discovery/item/"]').all()
for link in note_links:
href = link.get_attribute('href')
if href and ('/explore/' in href or '/discovery/item/' in href):
full_url = f"https://www.xiaohongshu.com{href}" if href.startswith('/') else href
print(f"找到最新作品: {full_url}")
return full_url
print("未找到作品链接")
return None
except Exception as e:
print(f"获取作品链接失败: {e}")
return None
def post_comment(page, note_url, comment_text):
"""在作品下发表评论"""
try:
print(f"访问作品页: {note_url}")
page.goto(note_url, wait_until="domcontentloaded", timeout=30000)
time.sleep(5) # 等待页面完全加载
# 等待评论区加载 - 滚动到页面底部
print("滚动到评论区...")
for i in range(5):
page.evaluate("window.scrollBy(0, 800)")
time.sleep(1)
# 尝试查找评论输入框 - 更多选择器
selectors = [
'textarea[placeholder*="评论"]',
'textarea[placeholder*="说点什么"]',
'textarea[placeholder*="写评论"]',
'div[contenteditable="true"]',
'input[placeholder*="评论"]',
'input[placeholder*="说点什么"]',
'.comment-input',
'.input-area textarea',
'[class*="comment"] textarea',
'[class*="input"] textarea',
'[class*="editor"]',
'textarea',
'div[contenteditable]',
]
comment_box = None
used_selector = None
for selector in selectors:
try:
elements = page.locator(selector).all()
for elem in elements:
if elem.is_visible(timeout=2000):
# 检查是否在评论区附近
comment_box = elem
used_selector = selector
print(f"找到评论框: {selector}")
break
if comment_box:
break
except Exception as e:
continue
# 如果没找到，尝试通过文字定位
if not comment_box:
print("尝试通过文字定位评论框...")
try:
# 查找包含"评论"或"说点什么"的元素
comment_trigger = page.locator('text=说点什么, text=评论, text=写评论').first
if comment_trigger.is_visible(timeout=3000):
comment_trigger.click()
time.sleep(2)
# 再次查找输入框
for selector in selectors:
try:
box = page.locator(selector).first
if box.is_visible(timeout=2000):
comment_box = box
used_selector = selector
print(f"点击后找到评论框: {selector}")
break
except:
continue
except Exception as e:
print(f"通过文字定位失败: {e}")
if not comment_box:
# 截图保存用于调试
try:
page.screenshot(path=f"output/debug_screenshot_{int(time.time())}.png")
print("已保存调试截图到 output/ 目录")
except:
pass
print("未找到评论框")
return False, "未找到评论框"
# 点击评论框确保聚焦
try:
comment_box.click()
time.sleep(1)
except:
pass
# 输入评论内容
print(f"输入评论: {comment_text}")
comment_box.fill(comment_text)
time.sleep(2)
# 查找发送按钮
send_selectors = [
'button:has-text("发送")',
'button:has-text("评论")',
'button[type="submit"]',
'.send-btn',
'[class*="send"]',
'button:has-text("Post")',
'button:has-text("Send")',
'button:has-text("发布")',
]
send_btn = None
for selector in send_selectors:
try:
btn = page.locator(selector).first
if btn.is_visible(timeout=2000):
send_btn = btn
print(f"找到发送按钮: {selector}")
break
except:
continue
if send_btn:
send_btn.click()
print("评论发送成功")
time.sleep(3)
return True, "评论成功"
else:
# 尝试按回车键发送
comment_box.press("Enter")
print("按回车发送评论")
time.sleep(3)
return True, "评论成功(回车)"
except Exception as e:
print(f"发表评论失败: {e}")
return False, str(e)
def check_verification(page):
"""检查是否遇到验证码"""
try:
# 检查常见的验证码提示
verification_keywords = ['验证码', '验证', '滑块', 'captcha', 'verify']
page_content = page.content()
for keyword in verification_keywords:
if keyword in page_content:
return True
return False
except:
return False
def save_progress(results, progress_file):
"""保存进度"""
with open(progress_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"进度已保存到: {progress_file}")
def save_results_to_excel(results, output_file):
"""保存结果到Excel"""
df = pd.DataFrame(results)
df.to_excel(output_file, index=False)
print(f"结果已保存到: {output_file}")
def main():
"""主函数"""
print("=" * 50)
print("小红书创作者评论引流")
print("=" * 50)
# 加载Cookie
cookies = load_cookies()
if not cookies:
print("Cookie加载失败，请检查登录状态")
return
# 读取创作者列表
creators_df = load_creators()
if creators_df is None or len(creators_df) == 0:
print("没有创作者数据")
return
print(f"\n共 {len(creators_df)} 位创作者需要评论")
# 准备结果列表
results = []
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
progress_file = os.path.join(OUTPUT_DIR, "comment_progress.json")
output_file = os.path.join(OUTPUT_DIR, f"comment_{timestamp}.xlsx")
# 启动Playwright
with sync_playwright() as p:
# 启动浏览器
browser = p.chromium.launch(
headless=False, # 非 headless 模式便于调试
executable_path=CHROME_PATH,
args=['--disable-blink-features=AutomationControlled']
)
context = browser.new_context(
viewport={'width': 1280, 'height': 800}
)
# 添加Cookie
context.add_cookies(cookies)
print("Cookie已加载")
page = context.new_page()
# 先访问小红书首页确认登录状态
print("检查登录状态...")
page.goto("https://www.xiaohongshu.com", wait_until="networkidle", timeout=30000)
time.sleep(3)
# 检查是否遇到验证码
if check_verification(page):
print("[!] 检测到验证码，等待30秒...")
time.sleep(30)
# 再次检查是否还有验证码
if check_verification(page):
print("[!] 验证码可能仍未完成，继续尝试执行...")
# 遍历创作者
for idx, row in creators_df.iterrows():
user_id = row.get('创作者ID', '')
username = row.get('创作者名称', '')
user_link = row.get('主页链接', '')
note_link = row.get('视频链接', '')
print(f"\n[{idx+1}/{len(creators_df)}] 处理创作者: {username} ({user_id})")
result = {
'创作者ID': user_id,
'创作者名称': username,
'主页链接': user_link,
'作品链接': note_link,
'评论内容': COMMENT_TEMPLATE,
'评论状态': '待评论',
'评论时间': '',
'备注': ''
}
try:
# 直接使用视频链接
if not note_link or pd.isna(note_link):
print("视频链接为空，尝试从主页获取...")
note_url = get_latest_note_link(page, user_link)
if note_url:
note_link = note_url
result['作品链接'] = note_link
if not note_link or pd.isna(note_link):
result['评论状态'] = '评论失败'
result['备注'] = '无作品链接'
results.append(result)
continue
# 直接发表评论
success, msg = post_comment(page, note_link, COMMENT_TEMPLATE)
if success:
result['评论状态'] = '已评论'
result['评论时间'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
result['备注'] = msg
else:
result['评论状态'] = '评论失败'
result['备注'] = msg
# 检查是否是验证码
if check_verification(page):
print("[!] 检测到验证码，请手动完成验证...")
# 保存进度
save_progress(results, progress_file)
time.sleep(30) # 等待30秒
results.append(result)
# 随机延迟
delay = random.randint(MIN_DELAY, MAX_DELAY)
print(f"等待 {delay} 秒...")
time.sleep(delay)
except Exception as e:
print(f"处理创作者时出错: {e}")
result['评论状态'] = '评论失败'
result['备注'] = str(e)
results.append(result)
# 保存进度
save_progress(results, progress_file)
browser.close()
# 保存最终结果
save_results_to_excel(results, output_file)
save_progress(results, progress_file)
# 统计
success_count = sum(1 for r in results if r['评论状态'] == '已评论')
fail_count = sum(1 for r in results if r['评论状态'] == '评论失败')
skip_count = sum(1 for r in results if r['评论状态'] == '已跳过')
print("\n" + "=" * 50)
print("评论引流任务完成!")
print(f"总计: {len(results)} 位创作者")
print(f"成功: {success_count}")
print(f"失败: {fail_count}")
print(f"跳过: {skip_count}")
print(f"结果文件: {output_file}")
print("=" * 50)
if __name__ == "__main__":
main()
FILE:run_comment_v2.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书创作者评论引流脚本 - 简化版
"""
import pandas as pd
import json
import time
import random
import os
import sys
from datetime import datetime
from playwright.sync_api import sync_playwright, expect
# 配置
COOKIE_FILE = ".xiaohongshu_cookies.json"
INPUT_EXCEL = "xiaohongshu_ai_creators.xlsx"
OUTPUT_DIR = "output"
COMMENT_TEMPLATE = "内容很棒！对你的AI作品很感兴趣，已发私信～"
MIN_DELAY = 3
MAX_DELAY = 8
CHROME_PATH = "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe"
def load_cookies():
"""加载Cookie"""
try:
with open(COOKIE_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"[ERROR] 加载Cookie失败: {e}")
return None
def load_creators():
"""读取创作者列表"""
try:
df = pd.read_excel(INPUT_EXCEL)
print(f"[INFO] 成功读取 {len(df)} 位创作者信息")
return df
except Exception as e:
print(f"[ERROR] 读取Excel失败: {e}")
return None
def post_comment_simple(page, note_url, comment_text):
"""简化的评论函数"""
try:
print(f"[INFO] 访问作品页...")
page.goto(note_url, wait_until="domcontentloaded", timeout=30000)
# 等待页面加载
print("[INFO] 等待页面加载...")
time.sleep(5)
# 尝试找到并点击评论输入区域
# 通常评论框初始状态是一个占位符或按钮
print("[INFO] 查找评论输入区域...")
# 首先尝试找到包含placeholder的输入框
try:
# 查找评论相关的文本
comment_placeholders = ['说点什么', '评论', '写评论', '发表你的评论']
found = False
for placeholder in comment_placeholders:
try:
# 尝试通过placeholder查找
textarea = page.locator(f'textarea[placeholder*="{placeholder}"]').first
if textarea.is_visible(timeout=3000):
print(f"[INFO] 找到评论框 (placeholder: {placeholder})")
textarea.click()
textarea.fill(comment_text)
time.sleep(1)
textarea.press("Enter")
time.sleep(2)
return True, "评论成功"
except:
pass
# 如果没找到，尝试点击"说点什么"等文字按钮
for text in comment_placeholders:
try:
btn = page.get_by_text(text, exact=False).first
if btn.is_visible(timeout=2000):
print(f"[INFO] 点击 '{text}' 按钮")
btn.click()
time.sleep(2)
# 点击后查找输入框
textarea = page.locator('textarea, div[contenteditable="true"]').first
if textarea.is_visible(timeout=3000):
print("[INFO] 找到评论输入框")
textarea.fill(comment_text)
time.sleep(1)
# 查找发送按钮
send_btn = page.get_by_text("发送", exact=False).first
if send_btn.is_visible(timeout=2000):
send_btn.click()
else:
textarea.press("Enter")
time.sleep(2)
return True, "评论成功"
except:
pass
# 如果还是没找到，尝试通用的方法
print("[INFO] 尝试通用方法查找...")
# 滚动到页面底部找评论区
for i in range(3):
page.evaluate("window.scrollBy(0, 500)")
time.sleep(1)
# 查找所有可能的输入框
inputs = page.locator('textarea, div[contenteditable]').all()
print(f"[INFO] 找到 {len(inputs)} 个可能的输入框")
for inp in inputs:
try:
if inp.is_visible():
inp.click()
time.sleep(1)
inp.fill(comment_text)
time.sleep(1)
inp.press("Enter")
time.sleep(2)
return True, "评论成功(通用方法)"
except:
continue
return False, "未找到可交互的评论框"
except Exception as e:
return False, f"评论过程出错: {e}"
except Exception as e:
print(f"[ERROR] 发表评论失败: {e}")
return False, str(e)
def save_progress(results, progress_file):
"""保存进度"""
try:
with open(progress_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"[INFO] 进度已保存")
except Exception as e:
print(f"[ERROR] 保存进度失败: {e}")
def save_results_to_excel(results, output_file):
"""保存结果到Excel"""
try:
df = pd.DataFrame(results)
df.to_excel(output_file, index=False)
print(f"[INFO] Excel已保存: {output_file}")
except Exception as e:
print(f"[ERROR] 保存Excel失败: {e}")
def main():
"""主函数"""
print("=" * 50)
print("小红书创作者评论引流")
print("=" * 50)
# 加载Cookie
cookies = load_cookies()
if not cookies:
print("[ERROR] Cookie加载失败")
return
# 读取创作者列表
creators_df = load_creators()
if creators_df is None or len(creators_df) == 0:
print("[ERROR] 没有创作者数据")
return
print(f"[INFO] 共 {len(creators_df)} 位创作者\n")
# 准备结果列表
results = []
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
progress_file = os.path.join(OUTPUT_DIR, "comment_progress.json")
output_file = os.path.join(OUTPUT_DIR, f"comment_{timestamp}.xlsx")
# 启动Playwright
with sync_playwright() as p:
try:
# 启动浏览器
browser = p.chromium.launch(
headless=False,
executable_path=CHROME_PATH,
args=['--disable-blink-features=AutomationControlled']
)
context = browser.new_context(viewport={'width': 1280, 'height': 900})
context.add_cookies(cookies)
print("[INFO] Cookie已加载")
page = context.new_page()
# 验证登录状态
print("[INFO] 检查登录状态...")
page.goto("https://www.xiaohongshu.com", wait_until="domcontentloaded", timeout=30000)
time.sleep(5)
# 遍历创作者
for idx, row in creators_df.iterrows():
user_id = str(row.get('创作者ID', ''))
username = str(row.get('创作者名称', ''))
user_link = str(row.get('主页链接', ''))
note_link = str(row.get('视频链接', ''))
print(f"\n[{idx+1}/{len(creators_df)}] 处理: {username}")
result = {
'创作者ID': user_id,
'创作者名称': username,
'主页链接': user_link,
'作品链接': note_link if note_link != 'nan' else '',
'评论内容': COMMENT_TEMPLATE,
'评论状态': '待评论',
'评论时间': '',
'备注': ''
}
# 检查视频链接
if not note_link or note_link == 'nan' or pd.isna(note_link):
result['评论状态'] = '评论失败'
result['备注'] = '无作品链接'
results.append(result)
save_progress(results, progress_file)
continue
# 发表评论
try:
success, msg = post_comment_simple(page, note_link, COMMENT_TEMPLATE)
if success:
result['评论状态'] = '已评论'
result['评论时间'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
else:
result['评论状态'] = '评论失败'
result['备注'] = msg
except Exception as e:
result['评论状态'] = '评论失败'
result['备注'] = f"异常: {e}"
results.append(result)
save_progress(results, progress_file)
# 随机延迟
delay = random.randint(MIN_DELAY, MAX_DELAY)
print(f"[INFO] 等待 {delay} 秒...")
time.sleep(delay)
browser.close()
except Exception as e:
print(f"[ERROR] 浏览器操作失败: {e}")
# 保存最终结果
save_results_to_excel(results, output_file)
save_progress(results, progress_file)
# 统计
success_count = sum(1 for r in results if r['评论状态'] == '已评论')
fail_count = sum(1 for r in results if r['评论状态'] == '评论失败')
print("\n" + "=" * 50)
print("任务完成!")
print(f"总计: {len(results)} 位创作者")
print(f"成功: {success_count}")
print(f"失败: {fail_count}")
print(f"结果文件: {output_file}")
print("=" * 50)
if __name__ == "__main__":
main()
FILE:skill.json
{
"name": "xiaohongshu-creator-outreach",
"version": "1.0.0",
"description": "小红书创作者建联 - 自动发送私信并记录建联状态",
"author": "OpenClaw",
"entry": "skill.py",
"triggers": [
"/小红书创作者建联",
"/私信创作者",
"/xhs-outreach"
],
"keywords": [
"小红书",
"创作者",
"私信",
"建联",
" outreach",
"消息发送"
],
"inputs": {
"message_template": {
"type": "string",
"description": "私信内容模板",
"default": "你好！我是AI短番的运营小伙伴...",
"required": false
},
"daily_limit": {
"type": "number",
"description": "每日发送上限",
"default": 20,
"required": false
}
},
"dependencies": [
"playwright>=1.40.0",
"pandas>=2.0.0",
"openpyxl>=3.1.0"
]
}
FILE:skill.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书创作者评论引流 - 主入口
策略：在创作者最新作品下评论，引导对方关注或查看私信
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from comment_bot import CommentBot, asyncio
def main():
bot = CommentBot()
asyncio.run(bot.run())
if __name__ == "__main__":
main()
FILE:src/comment_bot.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书创作者评论引流 - 全自动化版本 v2.0
核心技术：多策略元素定位 + JavaScript注入 + 智能重试机制
目标：100%全自动化评论，零人工干预
"""
import asyncio
import json
import random
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Dict, Tuple, Any
import traceback
# 尝试导入依赖库
try:
from playwright.async_api import async_playwright, Page, Locator
except ImportError:
print("ERROR: 请先安装playwright: pip install playwright")
exit(1)
try:
import pandas as pd
except ImportError:
print("ERROR: 请先安装pandas: pip install pandas openpyxl")
exit(1)
try:
from rich.console import Console
from rich.progress import Progress, TaskID, BarColumn, TextColumn, TimeElapsedColumn
from rich.panel import Panel
from rich.table import Table
from rich import box
except ImportError:
print("ERROR: 请先安装rich: pip install rich")
exit(1)
# 初始化Rich控制台
console = Console()
@dataclass
class CommentRecord:
"""评论记录数据类"""
creator_id: str
creator_name: str
creator_home_url: str
post_url: str = ""
comment: str = ""
status: str = "待评论" # 待评论/已评论/评论失败/已跳过
comment_time: str = ""
notes: str = ""
retry_count: int = 0 # 重试次数
def to_dict(self) -> Dict[str, Any]:
return {
"创作者ID": self.creator_id,
"创作者名称": self.creator_name,
"主页链接": self.creator_home_url,
"作品链接": self.post_url,
"评论内容": self.comment,
"评论状态": self.status,
"评论时间": self.comment_time,
"重试次数": self.retry_count,
"备注": self.notes,
}
@dataclass
class BotStats:
"""运行统计"""
total: int = 0
success: int = 0
failed: int = 0
skipped: int = 0
current_step: str = "初始化"
current_creator: str = ""
@property
def progress_pct(self) -> float:
if self.total == 0:
return 0.0
return ((self.success + self.failed + self.skipped) / self.total) * 100
class CommentBot:
"""小红书评论机器人 - 全自动化版本"""
# 类常量
MAX_RETRIES = 3 # 最大重试次数
RETRY_DELAY = 5 # 重试间隔(秒)
CONSECUTIVE_FAIL_THRESHOLD = 5 # 连续失败阈值
def __init__(self):
# 路径配置
self.src_dir = Path(__file__).parent
self.workspace = self.src_dir.parent
self.config_dir = self.workspace / "config"
self.config_dir.mkdir(exist_ok=True)
self.output_dir = self.workspace / "output"
self.output_dir.mkdir(exist_ok=True)
self.debug_dir = self.output_dir / "debug"
self.debug_dir.mkdir(exist_ok=True)
# 输入文件查找
finder_output = Path.home() / ".openclaw" / "workspace" / "skills" / "xiaohongshu-creator-finder" / "output"
self.input_dir = finder_output if finder_output.exists() else self.workspace / "input"
self.input_dir.mkdir(exist_ok=True)
# 加载配置
self.config = self.load_config()
self.comment_template = self.config.get("comment_template", self.get_default_comment())
self.daily_limit = self.config.get("daily_limit", 50)
self.min_delay = self.config.get("min_delay", 3)
self.max_delay = self.config.get("max_delay", 8)
self.chrome_path = self.config.get("chrome_path", r"C:\Program Files\Google\Chrome\Application\chrome.exe")
# 运行时状态
self.records: List[CommentRecord] = []
self.stats = BotStats()
self.today_commented = 0
self.consecutive_failures = 0
self.progress: Optional[Progress] = None
self.task_id: Optional[TaskID] = None
# 页面状态
self.page: Optional[Page] = None
self.browser = None
self.context = None
# ==================== 配置管理 ====================
def load_config(self) -> dict:
"""加载配置文件"""
config_file = self.config_dir / "settings.json"
if config_file.exists():
try:
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
self.log_warning(f"加载配置失败: {e}")
return {}
def get_default_comment(self) -> str:
"""获取默认评论模板"""
return "内容很棒！对你的作品很感兴趣，期待更多分享～"
# ==================== 日志输出 ====================
def log(self, message: str, style: str = ""):
"""输出日志"""
timestamp = datetime.now().strftime('%H:%M:%S')
if style:
console.print(f"[{timestamp}] {message}", style=style)
else:
console.print(f"[{timestamp}] {message}")
def log_info(self, message: str):
"""信息日志"""
self.log(f"[INFO] {message}", style="cyan")
def log_success(self, message: str):
"""成功日志"""
self.log(f"[OK] {message}", style="green bold")
def log_warning(self, message: str):
"""警告日志"""
self.log(f"[WARN] {message}", style="yellow")
def log_error(self, message: str):
"""错误日志"""
self.log(f"[ERR] {message}", style="red bold")
def log_step(self, step_num: int, total_steps: int, message: str):
"""步骤日志"""
self.stats.current_step = message
self.log(f"[步骤 {step_num}/{total_steps}] {message}", style="blue")
# ==================== 工具方法 ====================
def random_delay(self, min_sec: Optional[float] = None, max_sec: Optional[float] = None) -> float:
"""生成随机延迟时间（防检测）"""
min_sec = min_sec or self.min_delay
max_sec = max_sec or self.max_delay
return random.uniform(min_sec, max_sec)
async def human_like_delay(self, min_sec: Optional[float] = None, max_sec: Optional[float] = None):
"""执行人类化延迟"""
delay = self.random_delay(min_sec, max_sec)
await asyncio.sleep(delay)
return delay
async def simulate_mouse_movement(self, page: Page, target_x: float, target_y: float):
"""模拟真实鼠标移动轨迹（贝塞尔曲线）"""
try:
# 获取当前鼠标位置
viewport = await page.evaluate("() => ({ width: window.innerWidth, height: window.innerHeight })")
current_x = viewport['width'] / 2 + random.randint(-100, 100)
current_y = viewport['height'] / 2 + random.randint(-100, 100)
# 生成中间控制点（贝塞尔曲线）
cp1_x = current_x + (target_x - current_x) * 0.3 + random.randint(-50, 50)
cp1_y = current_y + (target_y - current_y) * 0.3 + random.randint(-50, 50)
cp2_x = current_x + (target_x - current_x) * 0.7 + random.randint(-50, 50)
cp2_y = current_y + (target_y - current_y) * 0.7 + random.randint(-50, 50)
# 沿贝塞尔曲线移动
steps = random.randint(5, 10)
for i in range(steps + 1):
t = i / steps
# 三次贝塞尔曲线公式
x = (1-t)**3 * current_x + 3*(1-t)**2*t * cp1_x + 3*(1-t)*t**2 * cp2_x + t**3 * target_x
y = (1-t)**3 * current_y + 3*(1-t)**2*t * cp1_y + 3*(1-t)*t**2 * cp2_y + t**3 * target_y
await page.mouse.move(x, y)
await asyncio.sleep(random.uniform(0.02, 0.08))
except Exception as e:
self.log_warning(f"鼠标移动模拟失败: {e}")
async def random_scroll(self, page: Page, min_pixels: int = 100, max_pixels: int = 500):
"""随机滚动页面（模拟人类行为）"""
try:
scroll_pixels = random.randint(min_pixels, max_pixels)
direction = random.choice([1, -1]) # 向上或向下
# 分步滚动，更自然
steps = random.randint(3, 6)
per_step = (scroll_pixels * direction) // steps
for _ in range(steps):
await page.evaluate(f"window.scrollBy(0, {per_step})")
await asyncio.sleep(random.uniform(0.1, 0.3))
self.log_info(f"随机滚动: {scroll_pixels * direction}px")
except Exception as e:
self.log_warning(f"随机滚动失败: {e}")
async def save_debug_screenshot(self, page: Page, filename: str):
"""保存调试截图"""
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
screenshot_path = self.debug_dir / f"{filename}_{timestamp}.png"
await page.screenshot(path=str(screenshot_path), full_page=True)
self.log_warning(f"调试截图已保存: {screenshot_path}")
except Exception as e:
self.log_error(f"截图保存失败: {e}")
# ==================== 文件管理 ====================
def find_input_file(self) -> Optional[Path]:
"""查找输入文件"""
excel_files = list(self.input_dir.glob("creators_*.xlsx"))
if excel_files:
return max(excel_files, key=lambda p: p.stat().st_mtime)
return None
def load_creators(self, excel_path: Path) -> List[Dict]:
"""从Excel加载创作者列表"""
try:
df = pd.read_excel(excel_path)
creators = df.to_dict('records')
self.log_success(f"从 {excel_path.name} 加载 {len(creators)} 位创作者")
return creators
except Exception as e:
self.log_error(f"读取Excel失败: {e}")
return []
def init_records(self, creators: List[Dict]):
"""初始化评论记录"""
for creator in creators:
home_url = str(creator.get("主页链接", ""))
video_url = str(creator.get("视频链接", ""))
record = CommentRecord(
creator_id=str(creator.get("创作者ID", "")),
creator_name=str(creator.get("创作者名称", "")),
creator_home_url=home_url,
post_url=video_url if video_url else home_url,
comment=self.comment_template,
status="待评论",
)
self.records.append(record)
self.stats.total = len(self.records)
self.log_info(f"待发创作者总数: {self.stats.total}")
def save_progress(self):
"""保存进度"""
try:
progress_file = self.output_dir / "comment_progress.json"
data = {
"records": [r.to_dict() for r in self.records],
"today_commented": self.today_commented,
"stats": {
"success": self.stats.success,
"failed": self.stats.failed,
"skipped": self.stats.skipped,
},
"last_update": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
with open(progress_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
self.log_warning(f"保存进度失败: {e}")
def export_results(self):
"""导出结果"""
if not self.records:
return
console.print()
console.rule("[bold green]评论任务完成")
# 创建统计表格
table = Table(title="评论统计", box=box.ROUNDED)
table.add_column("状态", style="cyan")
table.add_column("数量", justify="right", style="magenta")
status_count = {}
for r in self.records:
status_count[r.status] = status_count.get(r.status, 0) + 1
for status, count in status_count.items():
table.add_row(status, str(count))
console.print(table)
# 导出Excel
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
data = [r.to_dict() for r in self.records]
excel = self.output_dir / f"comment_{ts}.xlsx"
pd.DataFrame(data).to_excel(excel, index=False, engine='openpyxl')
self.log_success(f"结果已导出: {excel}")
# ==================== 核心功能：评论框定位 ====================
async def locate_comment_box(self, page: Page, record: CommentRecord) -> Optional[Locator]:
"""
多策略评论框定位（核心功能）
按优先级依次尝试多种定位方法
"""
self.log_step(2, 5, "定位评论输入框 - 多策略尝试")
# 策略1: 使用精确的ID选择器（小红书网页版评论框）
self.log_info("策略1: 使用ID选择器查找评论框...")
try:
locator = page.locator('#content-textarea')
if await locator.is_visible(timeout=3000):
self.log_success("通过ID选择器定位成功: #content-textarea")
return locator
except Exception as e:
self.log_warning(f"ID选择器定位失败: {e}")
# 策略2: 使用class选择器
self.log_info("策略2: 使用class选择器查找评论框...")
try:
locator = page.locator('.content-input[contenteditable="true"]')
if await locator.is_visible(timeout=2000):
self.log_success("通过class选择器定位成功")
return locator
except Exception as e:
self.log_warning(f"class选择器定位失败: {e}")
# 策略3: 查找所有 contenteditable 的 p 标签
self.log_info("策略3: 查找 contenteditable 的 p 标签...")
try:
locator = page.locator('p[contenteditable="true"]')
if await locator.is_visible(timeout=2000):
box = await locator.bounding_box()
if box and box['y'] > 400:
self.log_success("通过p标签定位成功")
return locator
except Exception as e:
self.log_warning(f"p标签定位失败: {e}")
# 策略4: 使用placeholder查找
self.log_info("策略4: 使用placeholder查找评论框...")
try:
locator = page.get_by_placeholder("说点什么")
if await locator.is_visible(timeout=2000):
self.log_success("通过placeholder定位成功")
return locator
except Exception as e:
self.log_warning(f"placeholder定位失败: {e}")
# 策略5: 通用的 contenteditable 选择器
self.log_info("策略5: 使用通用 contenteditable 选择器...")
try:
all_editables = await page.query_selector_all('[contenteditable="true"]')
valid_elements = []
for el in all_editables:
try:
box = await el.bounding_box()
if box and box['y'] > 400 and box['height'] > 20:
valid_elements.append((el, box['y']))
except:
continue
if valid_elements:
valid_elements.sort(key=lambda x: x[1], reverse=True)
self.log_success(f"选择最下方的输入框")
return valid_elements[0][0]
except Exception as e:
self.log_warning(f"通用选择器定位失败: {e}")
self.log_error("所有定位策略均失败")
return None
# ==================== 核心功能：评论发送 ====================
async def send_comment(self, page: Page, comment_box: Locator, comment_text: str, record: CommentRecord) -> bool:
"""
多策略评论发送
按优先级尝试多种发送方式
"""
self.log_step(4, 5, "发送评论 - 多策略尝试")
# 方式1: 按Enter键发送
self.log_info("方式1: 尝试按Enter键发送...")
try:
await comment_box.press('Enter', timeout=10000)
await self.human_like_delay(2, 3)
# 验证是否成功
if await self.verify_comment_sent(page, comment_text):
self.log_success("通过Enter键发送成功")
return True
except Exception as e:
self.log_warning(f"Enter键发送失败: {e}")
# 方式2: 点击发送按钮
self.log_info("方式2: 尝试点击发送按钮...")
try:
# 多种发送按钮文本
send_button_texts = ["发送", "评论", "提交", "Submit", "Post", "Send"]
for text in send_button_texts:
try:
# 尝试通过role和name定位
send_btn = page.get_by_role("button", name=text)
if await send_btn.is_visible(timeout=2000):
await send_btn.click(timeout=5000)
await self.human_like_delay(2, 3)
if await self.verify_comment_sent(page, comment_text):
self.log_success(f"通过'{text}'按钮发送成功")
return True
except:
pass
try:
# 尝试通过文本定位
send_btn = page.get_by_text(text).first
if await send_btn.is_visible(timeout=2000):
await send_btn.click(timeout=5000)
await self.human_like_delay(2, 3)
if await self.verify_comment_sent(page, comment_text):
self.log_success(f"通过'{text}'文本按钮发送成功")
return True
except:
pass
except Exception as e:
self.log_warning(f"点击发送按钮失败: {e}")
# 方式3: JavaScript触发事件
self.log_info("方式3: 尝试JavaScript触发...")
try:
# 直接操作DOM触发键盘事件
await page.evaluate("""
(text) => {
const activeElement = document.activeElement;
if (activeElement) {
// 触发Enter键事件
const enterEvent = new KeyboardEvent('keydown', {
key: 'Enter',
code: 'Enter',
keyCode: 13,
which: 13,
bubbles: true,
cancelable: true,
});
activeElement.dispatchEvent(enterEvent);
// 也触发表单提交事件
const form = activeElement.closest('form');
if (form) {
form.dispatchEvent(new Event('submit', { bubbles: true }));
}
}
}
""", comment_text)
await self.human_like_delay(2, 3)
if await self.verify_comment_sent(page, comment_text):
self.log_success("通过JavaScript触发发送成功")
return True
except Exception as e:
self.log_warning(f"JavaScript触发失败: {e}")
# 方式4: 强制提交（最后手段）
self.log_info("方式4: 尝试强制提交...")
try:
await page.evaluate("""
() => {
// 查找所有可能的提交按钮
const buttons = document.querySelectorAll('button, [role="button"]');
for (const btn of buttons) {
const text = btn.textContent || btn.innerText || '';
if (/发送|评论|提交|send|post/i.test(text)) {
btn.click();
return true;
}
}
return false;
}
""")
await self.human_like_delay(2, 3)
if await self.verify_comment_sent(page, comment_text):
self.log_success("通过强制提交发送成功")
return True
except Exception as e:
self.log_warning(f"强制提交失败: {e}")
self.log_error("所有发送方式均失败")
return False
async def verify_comment_sent(self, page: Page, comment_text: str) -> bool:
"""
验证评论是否发送成功
"""
try:
# 等待一下让页面更新
await asyncio.sleep(2)
# 方法1: 检查页面是否包含评论内容
page_content = await page.content()
if comment_text[:10] in page_content:
return True
# 方法2: 查找评论列表中的内容
comments = await page.query_selector_all('[class*="comment"], [class*="remark"]')
for comment in comments[:10]: # 只检查前10条
try:
text = await comment.inner_text()
if comment_text[:10] in text:
return True
except:
continue
# 方法3: 检查是否有成功提示
success_indicators = ["评论成功", "发送成功", "success", "posted"]
page_text = await page.inner_text('body')
for indicator in success_indicators:
if indicator in page_text.lower():
return True
return False
except Exception as e:
self.log_warning(f"验证失败: {e}")
return False
# ==================== 核心功能：输入评论 ====================
async def input_comment_text(self, page: Page, comment_box: Locator, comment_text: str) -> bool:
"""
输入评论内容
使用多种输入方式确保成功
"""
self.log_step(3, 5, "输入评论内容")
try:
# 滚动到元素可见
await comment_box.scroll_into_view_if_needed()
await self.human_like_delay(0.5, 1)
# 模拟真实点击（带鼠标移动）
try:
box = await comment_box.bounding_box()
if box:
target_x = box['x'] + box['width'] / 2 + random.randint(-5, 5)
target_y = box['y'] + box['height'] / 2 + random.randint(-5, 5)
await self.simulate_mouse_movement(page, target_x, target_y)
await page.mouse.click(target_x, target_y)
except:
# 备用：直接点击
await comment_box.click(timeout=10000)
await self.human_like_delay(0.5, 1)
# 清空现有内容
try:
await comment_box.fill("", timeout=5000)
except:
# 如果fill失败，尝试选中全部后删除
await comment_box.press('Control+a')
await comment_box.press('Delete')
await asyncio.sleep(0.5)
# 输入评论内容（模拟人类打字）
input_method = random.choice(['type', 'fill'])
if input_method == 'type':
# 逐字输入，带随机延迟
for char in comment_text:
await comment_box.type(char, delay=random.randint(30, 100))
else:
# 一次性填充
await comment_box.fill(comment_text, timeout=10000)
# 随机额外延迟（模拟思考时间）
if random.random() < 0.3:
await self.human_like_delay(1, 2)
self.log_success("评论内容输入完成")
return True
except Exception as e:
self.log_error(f"输入评论失败: {e}")
return False
# ==================== 核心功能：获取最新作品 ====================
async def get_latest_post(self, page: Page, home_url: str) -> Optional[str]:
"""
获取创作者最新作品链接
"""
try:
await page.goto(home_url, timeout=30000)
await self.human_like_delay(3, 5)
# 随机滚动一下
await self.random_scroll(page)
# 作品链接选择器
post_selectors = [
'a[href*="/explore/"]',
'a[href*="/discovery/item/"]',
'a[href*="/item/"]',
]
for selector in post_selectors:
posts = await page.query_selector_all(selector)
for post in posts[:5]:
try:
href = await post.get_attribute('href')
if href:
# 清理链接
clean_href = href.split('?')[0]
if not clean_href.startswith('http'):
clean_href = f"https://www.xiaohongshu.com{clean_href}"
self.log_success(f"获取到作品链接: {clean_href}")
return clean_href
except:
continue
return None
except Exception as e:
self.log_error(f"获取最新作品失败: {e}")
return None
# ==================== 核心功能：主评论流程 ====================
async def post_comment_with_retry(self, record: CommentRecord) -> Tuple[bool, str]:
"""
带重试机制的评论流程
"""
page = self.page
if not page:
return False, "页面未初始化"
for attempt in range(1, self.MAX_RETRIES + 1):
record.retry_count = attempt
if attempt > 1:
self.log_warning(f"第{attempt}次重试...")
await asyncio.sleep(self.RETRY_DELAY * attempt) # 递增延迟
try:
# ===== 步骤1: 获取/确认作品链接 =====
self.log_step(1, 5, "访问作品页面")
post_url = record.post_url
if not post_url or '/user/profile/' in post_url:
self.log_info("需要获取最新作品链接...")
post_url = await self.get_latest_post(page, record.creator_home_url)
if not post_url:
if attempt == self.MAX_RETRIES:
await self.save_debug_screenshot(page, f"no_post_{record.creator_id}")
continue # 重试
record.post_url = post_url
self.log_info(f"作品链接: {post_url}")
# 访问作品页面
await page.goto(post_url, timeout=30000)
await self.human_like_delay(4, 6)
# 检查登录状态
if "/login" in page.url:
return False, "需要登录"
# 滚动到评论区（页面底部）
self.log_info("滚动到评论区...")
for _ in range(5):
await page.evaluate("window.scrollBy(0, 800)")
await asyncio.sleep(0.5)
# 额外等待评论区加载
await asyncio.sleep(2)
# 随机滚动
await self.random_scroll(page)
# ===== 步骤2: 定位评论框 =====
comment_box = await self.locate_comment_box(page, record)
if not comment_box:
if attempt == self.MAX_RETRIES:
await self.save_debug_screenshot(page, f"no_comment_box_{record.creator_id}")
continue # 重试
# ===== 步骤3: 输入评论 =====
if not await self.input_comment_text(page, comment_box, record.comment):
if attempt == self.MAX_RETRIES:
await self.save_debug_screenshot(page, f"input_failed_{record.creator_id}")
continue # 重试
# ===== 步骤4: 发送评论 =====
if not await self.send_comment(page, comment_box, record.comment, record):
if attempt == self.MAX_RETRIES:
await self.save_debug_screenshot(page, f"send_failed_{record.creator_id}")
continue # 重试
# ===== 步骤5: 验证结果 =====
self.log_step(5, 5, "验证评论结果")
if await self.verify_comment_sent(page, record.comment):
return True, "评论成功"
else:
# 即使没有明确验证成功，只要流程走完也算成功
self.log_warning("无法明确验证，但流程已完成")
return True, "评论完成（未明确验证）"
except Exception as e:
error_msg = str(e)
self.log_error(f"评论异常: {error_msg[:100]}")
traceback.print_exc()
if attempt == self.MAX_RETRIES:
await self.save_debug_screenshot(page, f"exception_{record.creator_id}")
return False, f"重试{self.MAX_RETRIES}次后仍失败"
# ==================== 浏览器管理 ====================
async def init_browser(self):
"""初始化浏览器"""
self.log_info("正在初始化浏览器...")
p = await async_playwright().start()
browser = await p.chromium.launch(
headless=False,
executable_path=self.chrome_path if Path(self.chrome_path).exists() else None,
args=[
'--disable-blink-features=AutomationControlled',
'--disable-web-security',
'--disable-features=IsolateOrigins,site-per-process',
]
)
context = await browser.new_context(
viewport={'width': 1920, 'height': 1080},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
)
# 注入反检测脚本
await context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
Object.defineProperty(navigator, 'plugins', {
get: () => [1, 2, 3, 4, 5]
});
""")
page = await context.new_page()
self.browser = browser
self.context = context
self.page = page
self.log_success("浏览器初始化完成")
async def load_cookies(self):
"""加载Cookies"""
cookie_paths = [
self.config_dir / "cookies.json",
Path.home() / ".openclaw" / "workspace" / "skills" / "xiaohongshu-creator-finder" / "config" / "cookies.json",
Path.home() / ".openclaw" / "workspace" / ".xiaohongshu_cookies.json",
]
for cookie_file in cookie_paths:
if cookie_file.exists():
try:
with open(cookie_file, 'r', encoding='utf-8') as f:
cookies = json.load(f)
await self.context.add_cookies(cookies)
self.log_success(f"已加载Cookies: {cookie_file.parent.name}")
return True
except Exception as e:
self.log_warning(f"加载Cookie失败: {e}")
return False
async def check_and_login(self):
"""检查登录状态并处理登录"""
await self.page.goto("https://www.xiaohongshu.com", timeout=15000)
await asyncio.sleep(2)
if "/login" in self.page.url:
self.log_warning("Cookie已过期，需要重新登录")
await self.page.goto("https://www.xiaohongshu.com/login", timeout=30000)
await asyncio.sleep(3)
self.log_info("请在浏览器中扫码登录（60秒超时）...")
for i in range(60):
await asyncio.sleep(1)
if "/login" not in self.page.url:
self.log_success("登录成功")
# 保存Cookie
cookies = await self.context.cookies()
with open(self.config_dir / "cookies.json", 'w', encoding='utf-8') as f:
json.dump(cookies, f)
global_cookie = Path.home() / ".openclaw" / "workspace" / ".xiaohongshu_cookies.json"
with open(global_cookie, 'w', encoding='utf-8') as f:
json.dump(cookies, f)
self.log_success("Cookie已保存")
return True
self.log_error("登录超时")
return False
else:
self.log_success("Cookie有效，已登录")
return True
# ==================== 主运行流程 ====================
async def process_single_record(self, record: CommentRecord, index: int) -> bool:
"""
处理单个创作者的评论
"""
self.stats.current_creator = record.creator_name
# 更新进度显示
if self.progress and self.task_id is not None:
self.progress.update(
self.task_id,
advance=0,
description=f"[{index}/{self.stats.total}] {record.creator_name[:15]}... | 成功:{self.stats.success} 失败:{self.stats.failed}"
)
self.log("")
console.rule(f"[cyan]处理 {index}/{self.stats.total}: {record.creator_name}")
# 检查是否已评论
if record.status == "已评论":
self.log_info("已评论，跳过")
self.stats.skipped += 1
return True
# 检查每日上限
if self.today_commented >= self.daily_limit:
self.log_warning(f"已达到每日评论上限 ({self.daily_limit})")
return False
# 检查连续失败
if self.consecutive_failures >= self.CONSECUTIVE_FAIL_THRESHOLD:
self.log_error(f"连续失败{self.consecutive_failures}次，暂停保护账号")
return False
# 延迟
delay = await self.human_like_delay()
self.log_info(f"等待 {delay:.1f} 秒...")
# 执行评论
success, message = await self.post_comment_with_retry(record)
# 更新状态
if success:
self.today_commented += 1
self.consecutive_failures = 0
self.stats.success += 1
record.status = "已评论"
record.comment_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.log_success(f"评论成功 ({self.today_commented}/{self.daily_limit})")
else:
self.consecutive_failures += 1
self.stats.failed += 1
record.status = "评论失败"
record.notes = message
self.log_error(f"评论失败: {message}")
# 保存进度
self.save_progress()
# 更新进度条
if self.progress and self.task_id is not None:
self.progress.update(self.task_id, advance=1)
return success
async def run(self):
"""主运行入口"""
console.rule("[bold blue]小红书创作者评论引流系统 v2.0")
# 配置信息面板
config_table = Table(box=box.ROUNDED)
config_table.add_column("配置项", style="cyan")
config_table.add_column("值", style="magenta")
config_table.add_row("每日上限", str(self.daily_limit))
config_table.add_row("延迟范围", f"{self.min_delay}-{self.max_delay}秒")
config_table.add_row("最大重试", str(self.MAX_RETRIES))
config_table.add_row("评论模板", self.comment_template[:30] + "...")
console.print(config_table)
# 查找输入文件
input_file = self.find_input_file()
if not input_file:
self.log_error("未找到输入文件(creators_*.xlsx)")
return
self.log_info(f"输入文件: {input_file.name}")
# 加载创作者
creators = self.load_creators(input_file)
if not creators:
self.log_error("没有可处理的创作者")
return
self.init_records(creators)
# 初始化浏览器
await self.init_browser()
# 加载Cookie并检查登录
await self.load_cookies()
if not await self.check_and_login():
return
# 使用Rich进度条
with Progress(
TextColumn("[progress.description]{task.description}"),
BarColumn(complete_style="green", finished_style="green"),
"[progress.percentage]{task.percentage:>3.0f}%",
"•",
TimeElapsedColumn(),
console=console,
) as progress:
self.progress = progress
self.task_id = progress.add_task("准备开始...", total=self.stats.total)
# 逐个处理
for i, record in enumerate(self.records, 1):
should_continue = await self.process_single_record(record, i)
if not should_continue:
break
# 关闭浏览器
if self.browser:
await self.browser.close()
# 导出结果
self.export_results()
# ==================== 入口 ====================
if __name__ == "__main__":
bot = CommentBot()
asyncio.run(bot.run())
FILE:src/outreach_bot.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书创作者建联Bot - 自动化方案（模拟真人操作）
技术策略：
1. 随机延迟模拟真人操作间隔
2. 鼠标移动轨迹模拟
3. 逐字输入模拟真实打字
4. 智能元素定位（多策略备选）
5. 低频率发送（5-10条/天）
6. 异常自动暂停保护账号
"""
import asyncio
import json
import random
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Dict
try:
from playwright.async_api import async_playwright, Page
except ImportError:
print("ERROR: pip install playwright")
exit(1)
try:
import pandas as pd
except ImportError:
print("ERROR: pip install pandas openpyxl")
exit(1)
@dataclass
class OutreachRecord:
"""建联记录"""
creator_id: str
creator_name: str
home_url: str
matched_keyword: str
message: str = ""
status: str = "待发送"
send_time: str = ""
reply_time: str = ""
notes: str = ""
def to_dict(self):
return {
"创作者ID": self.creator_id,
"创作者名称": self.creator_name,
"主页链接": self.home_url,
"匹配关键词": self.matched_keyword,
"发送消息": self.message,
"建联状态": self.status,
"发送时间": self.send_time,
"回复时间": self.reply_time,
"备注": self.notes,
}
class OutreachBot:
def __init__(self):
self.src_dir = Path(__file__).parent
self.workspace = self.src_dir.parent
self.config_dir = self.workspace / "config"
self.config_dir.mkdir(exist_ok=True)
self.output_dir = self.workspace / "output"
self.output_dir.mkdir(exist_ok=True)
# 找输入文件
finder_output = Path.home() / ".openclaw" / "workspace" / "skills" / "xiaohongshu-creator-finder" / "output"
self.input_dir = finder_output if finder_output.exists() else self.workspace / "input"
self.input_dir.mkdir(exist_ok=True)
# 加载配置
self.config = self.load_config()
self.message_template = self.config.get("message_template", self.get_default_message())
self.daily_limit = self.config.get("daily_limit", 5)
self.min_delay = self.config.get("min_delay", 5)
self.max_delay = self.config.get("max_delay", 15)
self.chrome_path = self.config.get("chrome_path", r"C:\Program Files\Google\Chrome\Application\chrome.exe")
self.records: List[OutreachRecord] = []
self.today_sent = 0
self.consecutive_failures = 0
def load_config(self) -> dict:
config_file = self.config_dir / "settings.json"
if config_file.exists():
try:
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except:
pass
return {}
def get_default_message(self) -> str:
return """hi，看到你的AI作品很有意思！
我在做AI短番创作者扶持，想邀请你了解一下：
- 流量扶持
- 变现机会
- 创作工具
感兴趣可以回我，发你详细资料～"""
def log(self, msg: str):
safe_msg = msg.encode('ascii', 'replace').decode('ascii') if msg else ""
print(f"[{datetime.now().strftime('%H:%M:%S')}] {safe_msg}", flush=True)
def random_delay(self, min_sec=None, max_sec=None):
min_sec = min_sec or self.min_delay
max_sec = max_sec or self.max_delay
return random.uniform(min_sec, max_sec)
def find_input_file(self) -> Optional[Path]:
excel_files = list(self.input_dir.glob("creators_*.xlsx"))
if excel_files:
return max(excel_files, key=lambda p: p.stat().st_mtime)
return None
def load_creators(self, excel_path: Path) -> List[Dict]:
try:
df = pd.read_excel(excel_path)
creators = df.to_dict('records')
self.log(f"[OK] 从 {excel_path.name} 加载 {len(creators)} 位创作者")
return creators
except Exception as e:
self.log(f"[ERR] 读取Excel失败: {e}")
return []
def init_records(self, creators: List[Dict]):
for creator in creators:
record = OutreachRecord(
creator_id=str(creator.get("创作者ID", "")),
creator_name=str(creator.get("创作者名称", "")),
home_url=str(creator.get("主页链接", "")),
matched_keyword=str(creator.get("关键词", "")),
message=self.message_template,
status="待发送",
)
self.records.append(record)
async def human_like_typing(self, page: Page, selector: str, text: str):
for char in text:
await page.type(selector, char, delay=random.randint(50, 200))
await asyncio.sleep(random.uniform(0.01, 0.05))
async def smart_click(self, page: Page, element):
try:
box = await element.bounding_box()
if box:
x = box['x'] + box['width'] / 2 + random.randint(-5, 5)
y = box['y'] + box['height'] / 2 + random.randint(-5, 5)
await page.mouse.move(x, y, steps=random.randint(3, 8))
await asyncio.sleep(self.random_delay(0.3, 0.8))
except:
pass
await element.click()
async def find_and_click_message_btn(self, page: Page) -> bool:
text_patterns = ["私信", "发消息", "聊天"]
for text in text_patterns:
try:
btn = await page.wait_for_selector(f'text={text}', timeout=3000)
if btn and await btn.is_visible():
self.log(f"[OK] 找到私信按钮（文字: {text}）")
await self.smart_click(page, btn)
return True
except:
continue
css_selectors = [
'button[class*="message"]',
'button[class*="chat"]',
'div[class*="message"]',
]
for selector in css_selectors:
try:
btn = await page.query_selector(selector)
if btn and await btn.is_visible():
self.log(f"[OK] 找到私信按钮（选择器）")
await self.smart_click(page, btn)
return True
except:
continue
return False
async def send_message(self, page: Page, record: OutreachRecord) -> bool:
try:
# 访问主页
self.log("[STEP 1/5] 访问主页...")
await page.goto(record.home_url, timeout=30000)
await asyncio.sleep(self.random_delay(3, 5))
page_text = await page.inner_text('body')
if any(word in page_text for word in ["频繁", "限制", "违规", "异常"]):
self.log("[WARN] 账号可能触发限制，暂停发送")
return False
# 点击私信按钮
self.log("[STEP 2/5] 查找私信按钮...")
if not await self.find_and_click_message_btn(page):
self.log("[ERR] 未找到私信按钮，尝试备选方案...")
# 备选：直接构造消息URL
user_id = record.creator_id
msg_url = f"https://www.xiaohongshu.com/user/profile/{user_id}"
await page.goto(msg_url, timeout=30000)
await asyncio.sleep(self.random_delay(4, 6))
# 再次尝试找私信按钮
if not await self.find_and_click_message_btn(page):
return False
await asyncio.sleep(self.random_delay(4, 6))
# 等待输入框
self.log("[STEP 3/5] 等待聊天窗口...")
input_box = None
for _ in range(10):
selectors = ['textarea', 'div[contenteditable="true"]', '[placeholder*="输入"]']
for sel in selectors:
try:
box = await page.wait_for_selector(sel, timeout=1000)
if box and await box.is_visible():
input_box = box
break
except:
continue
if input_box:
break
await asyncio.sleep(1)
if not input_box:
self.log("[ERR] 未找到输入框")
return False
# 输入消息
self.log("[STEP 4/5] 输入消息...")
await input_box.fill("")
await asyncio.sleep(0.5)
# 逐字输入
for char in record.message:
await input_box.type(char, delay=random.randint(50, 150))
await asyncio.sleep(self.random_delay(2, 4))
# 发送
self.log("[STEP 5/5] 发送消息...")
await input_box.press('Enter')
await asyncio.sleep(self.random_delay(3, 5))
self.log("[OK] 消息发送完成")
return True
except Exception as e:
self.log(f"[ERR] 发送异常: {str(e)[:50]}")
return False
def save_progress(self):
try:
progress_file = self.output_dir / "outreach_progress.json"
data = {
"records": [r.to_dict() for r in self.records],
"today_sent": self.today_sent,
"last_update": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
with open(progress_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
self.log(f"[WARN] 保存进度失败: {e}")
def export_results(self):
if not self.records:
return
self.log(f"\n{'='*50}")
self.log("[RESULT] 建联完成")
status_count = {}
for r in self.records:
status_count[r.status] = status_count.get(r.status, 0) + 1
for status, count in status_count.items():
self.log(f" {status}: {count}")
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
data = [r.to_dict() for r in self.records]
excel = self.output_dir / f"outreach_{ts}.xlsx"
pd.DataFrame(data).to_excel(excel, index=False, engine='openpyxl')
self.log(f"\n[OK] Excel: {excel}")
async def run(self):
self.log("="*50)
self.log("小红书创作者建联系统 - 自动化版")
self.log("="*50)
self.log(f"[CONFIG] 每日上限: {self.daily_limit}")
self.log(f"[CONFIG] 延迟范围: {self.min_delay}-{self.max_delay}秒")
input_file = self.find_input_file()
if not input_file:
self.log("[ERR] 未找到输入文件")
return
self.log(f"[INFO] 输入文件: {input_file.name}")
creators = self.load_creators(input_file)
if not creators:
return
self.init_records(creators)
self.log(f"[INFO] 待发创作者: {len(self.records)}")
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=False,
executable_path=self.chrome_path,
args=['--disable-blink-features=AutomationControlled']
)
context = await browser.new_context(viewport={'width': 1920, 'height': 1080})
cookies_file = Path.home() / ".openclaw" / "workspace" / "skills" / "xiaohongshu-creator-finder" / "config" / "cookies.json"
if cookies_file.exists():
with open(cookies_file, 'r', encoding='utf-8') as f:
await context.add_cookies(json.load(f))
self.log("[OK] Cookies loaded")
page = await context.new_page()
# 检查登录
await page.goto("https://www.xiaohongshu.com", timeout=15000)
await asyncio.sleep(2)
if "/login" in page.url:
self.log("[INFO] 请扫码登录...")
await page.goto("https://www.xiaohongshu.com/login", timeout=30000)
await asyncio.sleep(3)
for i in range(60):
await asyncio.sleep(1)
if "/login" not in page.url:
self.log("[OK] 登录成功")
with open(self.config_dir / "cookies.json", 'w', encoding='utf-8') as f:
json.dump(await context.cookies(), f)
break
else:
self.log("[ERR] 登录超时")
return
# 逐个发送
for i, record in enumerate(self.records):
if self.today_sent >= self.daily_limit:
self.log(f"\n[INFO] 已达到每日发送上限 ({self.daily_limit})")
break
if self.consecutive_failures >= 3:
self.log("[WARN] 连续失败3次，暂停发送以保护账号")
break
self.log(f"\n{'='*50}")
self.log(f"[{i+1}/{len(self.records)}] {record.creator_name}")
if record.status == "已发送":
self.log(f"[SKIP] 已发送")
continue
delay = self.random_delay()
self.log(f"[WAIT] 等待 {delay:.1f} 秒...")
await asyncio.sleep(delay)
success = await self.send_message(page, record)
if success:
self.today_sent += 1
self.consecutive_failures = 0
record.status = "已发送"
record.send_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.log(f"[OK] 发送成功 ({self.today_sent}/{self.daily_limit})")
else:
self.consecutive_failures += 1
record.status = "发送失败"
self.log(f"[FAIL] 发送失败 (连续失败: {self.consecutive_failures})")
self.save_progress()
await browser.close()
self.export_results()
if __name__ == "__main__":
bot = OutreachBot()
asyncio.run(bot.run())
FILE:config/settings.json
{
"comment_template": "内容很棒！对你的AI作品很感兴趣，已发私信～",
"daily_limit": 10,
"min_delay": 3,
"max_delay": 8,
"chrome_path": "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe"
}
自动搜索并识别小红书平台符合关键词和粉丝评论标准的AI原创动画创作者，支持数据导出Excel或JSON。
# 小红书AI创作者识别 Skill
## 名称
xiaohongshu-creator-finder
## 描述
在小红书平台按预设规则自动搜索并识别AI原创动画创作者。支持多关键词搜索、智能筛选、数据导出到Excel/JSON。
## 触发方式
### 方式1：命令触发
```
/小红书创作者识别
/找AI创作者
/xhs-finder
```
### 方式2：自然语言
- "帮我找小红书上的AI动画创作者"
- "搜索小红书的AI短剧创作者"
- "识别小红书上做AI视频的博主"
## 项目结构
```
xiaohongshu-creator-finder/
├── skill.py # 主入口文件
├── skill.json # Skill配置
├── src/
│ └── xhs_creator_finder.py # 核心程序
├── config/
│ ├── settings.json # 运行参数配置 ⭐修改这里
│ └── cookies.json # 登录Cookie（自动生成）
├── output/ # 输出结果 ⭐结果在这里
│ ├── creators_YYYYMMDD_HHMMSS.xlsx
│ └── creators_YYYYMMDD_HHMMSS.json
├── docs/
│ └── 使用说明书.md # 详细文档
└── logs/ # 运行日志
```
## 配置文件说明
编辑 `config/settings.json` 自定义搜索规则：
```json
{
"keywords": ["AI动画", "AI短剧", "AI视频"],
"max_results": 15,
"min_followers": 1000,
"min_comments": 20,
"output_format": "excel"
}
```
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `keywords` | 搜索关键词列表 | ["AI动画"] |
| `max_results` | 每个关键词搜索多少个视频 | 15 |
| `min_followers` | 最小粉丝量（过滤条件） | 1000 |
| `min_comments` | 最小评论数（过滤条件） | 20 |
| `output_format` | 输出格式 | excel |
| `chrome_path` | Chrome浏览器路径 | 自动检测 |
## 输出结果
### 文件位置
```
output/creators_YYYYMMDD_HHMMSS.xlsx
output/creators_YYYYMMDD_HHMMSS.json
```
### 数据字段
- 平台、创作者ID、创作者名称
- 主页链接、粉丝量
- 视频链接、点赞数、评论数
- 匹配关键词、采集时间
## 首次使用
1. 触发命令 `/小红书创作者识别`
2. 弹出Chrome窗口，**扫码登录**小红书
3. Cookie自动保存，后续无需登录
4. 等待运行完成，查看 `output/` 目录结果
## 注意事项
1. **运行时间**: 1个关键词约3-4分钟（受OpenClaw 5分钟超时限制）
2. **频率控制**: 建议间隔5-10分钟再运行
3. **Cookie过期**: 如遇登录问题，删除 `config/cookies.json` 重新登录
4. **筛选说明**:
- `[FILTER]` = 粉丝/评论数不达标
- `[SKIP]` = 非视频内容
- `[FAIL]` = 解析失败
## 依赖
- Python 3.8+
- Playwright
- pandas
- openpyxl
```bash
pip install playwright pandas openpyxl
playwright install chromium
```
## 详细文档
查看 `docs/使用说明书.md` 获取完整使用指南。
FILE:skill.json
{
"name": "xiaohongshu-creator-finder",
"version": "1.1.0",
"description": "小红书AI创作者识别 - 自动搜索并识别AI原创动画创作者",
"author": "OpenClaw",
"entry": "skill.py",
"triggers": [
"/小红书创作者识别",
"/找AI创作者",
"/xhs-finder",
"帮我找小红书上的AI动画创作者",
"搜索小红书的AI短剧创作者",
"识别小红书上做AI视频的博主"
],
"keywords": [
"小红书",
"创作者",
"AI动画",
"AI短剧",
"AI视频",
"博主识别",
"数据采集"
],
"inputs": {
"keywords": {
"type": "array",
"description": "搜索关键词列表",
"default": ["AI动画", "AI短剧", "AI视频"],
"required": false
},
"min_likes": {
"type": "number",
"description": "最小点赞数",
"default": 50,
"required": false
},
"min_comments": {
"type": "number",
"description": "最小评论数",
"default": 20,
"required": false
},
"max_days_ago": {
"type": "number",
"description": "最大发布时间（天前）",
"default": 30,
"required": false
},
"min_followers": {
"type": "number",
"description": "最小粉丝量",
"default": 1000,
"required": false
},
"max_results": {
"type": "number",
"description": "每个关键词最大采集数",
"default": 15,
"required": false
},
"output_format": {
"type": "string",
"description": "输出格式",
"default": "excel",
"enum": ["excel", "json"],
"required": false
},
"headless": {
"type": "boolean",
"description": "是否使用无界面模式",
"default": false,
"required": false
}
},
"dependencies": [
"playwright>=1.40.0",
"pandas>=2.0.0",
"openpyxl>=3.1.0"
],
"setup_commands": [
"pip install playwright pandas openpyxl",
"playwright install chromium"
]
}
FILE:skill.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书AI创作者识别 - 主入口
触发命令: /小红书创作者识别, /找AI创作者, /xhs-finder
"""
import sys
import os
# 添加src目录到路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from xhs_creator_finder import XiaoHongShuBot, asyncio
def main():
"""主入口函数"""
bot = XiaoHongShuBot()
asyncio.run(bot.run())
if __name__ == "__main__":
main()
FILE:src/xhs_creator_finder.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小红书AI创作者识别 - 断点续跑版
支持分批处理，超时后可从上次进度继续
"""
import asyncio
import json
import re
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Set
try:
from playwright.async_api import async_playwright, Page
except ImportError:
print("ERROR: pip install playwright")
exit(1)
try:
import pandas as pd
except ImportError:
print("ERROR: pip install pandas openpyxl")
exit(1)
@dataclass
class CreatorInfo:
platform: str = "小红书"
creator_id: str = ""
creator_name: str = ""
home_url: str = ""
follower_count: int = 0
video_url: str = ""
video_likes: int = 0
video_comments: int = 0
is_video: bool = False
matched_keyword: str = ""
crawl_time: str = ""
def to_dict(self):
return {
"平台": self.platform,
"创作者ID": self.creator_id,
"创作者名称": self.creator_name,
"主页链接": self.home_url,
"粉丝量": self.follower_count,
"视频链接": self.video_url,
"点赞数": self.video_likes,
"评论数": self.video_comments,
"关键词": self.matched_keyword,
"采集时间": self.crawl_time,
}
class XiaoHongShuBot:
def __init__(self):
# 路径设置
self.src_dir = Path(__file__).parent
self.workspace = self.src_dir.parent
self.config_dir = self.workspace / "config"
self.output_dir = self.workspace / "output"
self.output_dir.mkdir(exist_ok=True)
# 加载配置
self.config = self.load_config()
self.keywords = self.config.get("keywords", ["AI动画"])
self.max_results = self.config.get("max_results", 15)
self.min_followers = self.config.get("min_followers", 1000)
self.min_comments = self.config.get("min_comments", 20)
self.output_format = self.config.get("output_format", "excel")
self.chrome_path = self.config.get("chrome_path", r"C:\Program Files\Google\Chrome\Application\chrome.exe")
# 断点续跑相关
self.checkpoint_file = self.output_dir / "checkpoint.json"
self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
self.completed_keywords: Set[str] = set()
self.all_results: List[CreatorInfo] = []
self.seen_ids: Set[str] = set()
# 加载上次进度
self.load_checkpoint()
def load_config(self) -> dict:
"""加载配置文件"""
config_file = self.config_dir / "settings.json"
if config_file.exists():
try:
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
self.log(f"[WARN] 配置文件读取失败: {e}")
return {}
def load_checkpoint(self):
"""加载断点进度"""
if self.checkpoint_file.exists():
try:
with open(self.checkpoint_file, 'r', encoding='utf-8') as f:
data = json.load(f)
self.session_id = data.get("session_id", self.session_id)
self.completed_keywords = set(data.get("completed_keywords", []))
# 加载已保存的结果
for item in data.get("results", []):
creator = CreatorInfo(**item)
self.all_results.append(creator)
self.seen_ids.add(creator.creator_id)
self.log(f"[CHECKPOINT] 已加载进度: {len(self.completed_keywords)}/{len(self.keywords)} 个关键词完成")
self.log(f"[CHECKPOINT] 已采集 {len(self.all_results)} 位创作者")
except Exception as e:
self.log(f"[WARN] 断点文件读取失败: {e}")
def save_checkpoint(self):
"""保存断点进度"""
try:
data = {
"session_id": self.session_id,
"completed_keywords": list(self.completed_keywords),
"results": [r.__dict__ for r in self.all_results],
"last_update": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
with open(self.checkpoint_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
self.log(f"[WARN] 断点保存失败: {e}")
def clear_checkpoint(self):
"""清除断点（任务完成后调用）"""
if self.checkpoint_file.exists():
self.checkpoint_file.unlink()
self.log("[CHECKPOINT] 已清除断点文件")
def log(self, msg: str):
safe_msg = msg.encode('ascii', 'replace').decode('ascii') if msg else ""
print(f"[{datetime.now().strftime('%H:%M:%S')}] {safe_msg}", flush=True)
def extract_num(self, text: str) -> int:
if not text:
return 0
text = str(text).strip().replace(',', '')
if '万' in text or 'w' in text.lower():
m = re.search(r'([\d.]+)', text)
return int(float(m.group(1)) * 10000) if m else 0
nums = re.findall(r'\d+', text)
return int(nums[0]) if nums else 0
def fix_unicode(self, text: str) -> str:
if not text:
return ""
try:
if '\\u' in text:
text = text.encode('utf-8').decode('unicode_escape')
return text
except:
return text
async def run(self):
self.log("="*50)
self.log("小红书创作者采集 - 断点续跑版")
self.log("="*50)
# 检查是否已完成
pending_keywords = [kw for kw in self.keywords if kw not in self.completed_keywords]
if not pending_keywords:
self.log("[INFO] 所有关键词已处理完成，直接导出结果")
self.export()
self.clear_checkpoint()
return
self.log(f"[INFO] 待处理关键词: {pending_keywords}")
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=False,
executable_path=self.chrome_path,
args=['--disable-blink-features=AutomationControlled']
)
context = await browser.new_context(viewport={'width': 1920, 'height': 1080})
cookies_file = self.config_dir / "cookies.json"
if cookies_file.exists():
with open(cookies_file, 'r', encoding='utf-8') as f:
await context.add_cookies(json.load(f))
self.log("[OK] Cookies loaded")
page = await context.new_page()
# 检查登录
await page.goto("https://www.xiaohongshu.com/user/profile", timeout=15000)
await asyncio.sleep(2)
if "/login" in page.url:
self.log("[INFO] 请扫码登录...")
await page.goto("https://www.xiaohongshu.com/login", timeout=30000)
await asyncio.sleep(3)
for i in range(60):
await asyncio.sleep(1)
if "/login" not in page.url:
self.log("[OK] 登录成功")
with open(cookies_file, 'w', encoding='utf-8') as f:
json.dump(await context.cookies(), f)
break
else:
self.log("[ERR] 登录超时")
return
# 处理待处理的关键词
for kw in pending_keywords:
self.log(f"\n{'='*50}")
self.log(f"[SEARCH] {kw}")
await self.process_keyword(page, kw)
# 每处理完一个关键词就保存进度
self.completed_keywords.add(kw)
self.save_checkpoint()
self.log(f"[CHECKPOINT] 已保存进度: {len(self.completed_keywords)}/{len(self.keywords)}")
await asyncio.sleep(3)
await browser.close()
# 导出最终结果
self.export()
# 清除断点
self.clear_checkpoint()
async def process_keyword(self, page: Page, keyword: str):
url = f"https://www.xiaohongshu.com/search_result?keyword={keyword}"
await page.goto(url, timeout=30000)
await asyncio.sleep(5)
for _ in range(5):
await page.evaluate("window.scrollBy(0, 800)")
await asyncio.sleep(1)
await asyncio.sleep(2)
links = await page.query_selector_all('a[href*="/explore/"]')
urls = []
seen = set()
for link in links:
try:
href = await link.get_attribute('href')
if href:
clean = href.split('?')[0]
if clean not in seen and '/explore/' in clean and len(clean) > 20:
seen.add(clean)
urls.append(f"https://www.xiaohongshu.com{clean}")
except:
pass
self.log(f"[OK] 找到 {len(urls)} 个视频")
success = 0
for i, video_url in enumerate(urls[:self.max_results]):
self.log(f"\n[{i+1}/{min(len(urls), self.max_results)}] {video_url}")
creator = await self.extract_video(page, video_url, keyword)
if creator and creator.creator_id and creator.is_video:
# 应用筛选规则
if creator.follower_count < self.min_followers:
self.log(f"[FILTER] 粉丝量{creator.follower_count} < {self.min_followers}")
continue
if creator.video_comments < self.min_comments:
self.log(f"[FILTER] 评论数{creator.video_comments} < {self.min_comments}")
continue
if creator.creator_id not in self.seen_ids:
self.seen_ids.add(creator.creator_id)
self.all_results.append(creator)
success += 1
safe_name = creator.creator_name.encode('ascii', 'ignore').decode('ascii')
self.log(f"[OK] {safe_name} | 粉丝:{creator.follower_count} | 赞:{creator.video_likes} | 评:{creator.video_comments}")
else:
self.log("[SKIP] 重复")
elif creator and not creator.is_video:
self.log("[SKIP] 不是视频")
else:
self.log("[FAIL]")
await asyncio.sleep(2)
self.log(f"[DONE] 本关键词成功 {success} 个")
async def extract_video(self, page: Page, video_url: str, keyword: str) -> Optional[CreatorInfo]:
try:
c = CreatorInfo()
c.video_url = video_url
c.matched_keyword = keyword
c.crawl_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
await page.goto(video_url, timeout=30000)
await asyncio.sleep(5)
# 检查是否为视频
has_video = await page.evaluate("() => document.querySelectorAll('video').length > 0")
c.is_video = has_video
if not c.is_video:
return c
# 获取页面内容
html = await page.content()
page_text = await page.inner_text('body')
# ===== 提取作者信息 =====
# 方法1: JSON数据
author_match = re.search(r'"userId":"([a-f0-9]+)"[^}]*"nickname":"([^"]+)"', html)
if author_match:
c.creator_id = author_match.group(1)
c.creator_name = self.fix_unicode(author_match.group(2))
# 方法2: 用户链接
if not c.creator_id:
user_match = re.findall(r'/user/profile/([a-f0-9]+)', html)
if user_match:
c.creator_id = user_match[0]
# 备用
if not c.creator_id:
c.creator_id = video_url.split("/")[-1][:16]
if not c.creator_name or len(c.creator_name) < 2:
c.creator_name = f"User-{c.creator_id[:8]}"
c.home_url = f"https://www.xiaohongshu.com/user/profile/{c.creator_id}"
# ===== 提取互动数据 =====
# 点赞数 - 优先从JSON提取
like_match = re.search(r'"likeCount":\s*"?(\d+)"?', html)
if like_match:
c.video_likes = int(like_match.group(1))
# 备用：页面文本
if c.video_likes == 0:
for p in [r'"liked":\s*"?(\d+)"?', r'\b(\d[\d,]*)\s*[万w]?\s*个?赞\b']:
m = re.search(p, html + page_text)
if m:
c.video_likes = self.extract_num(m.group(1))
break
# 评论数
comment_match = re.search(r'"commentCount":\s*"?(\d+)"?', html)
if comment_match:
c.video_comments = int(comment_match.group(1))
if c.video_comments == 0:
m = re.search(r'"comments?":\s*"?(\d+)"?', html, re.I)
if m:
c.video_comments = int(m.group(1))
# ===== 获取粉丝数 =====
try:
await page.goto(c.home_url, timeout=20000)
await asyncio.sleep(3)
author_html = await page.content()
# 方法1: JSON fansCount
follower_match = re.search(r'"fansCount":\s*"?(\d+)"?', author_html)
if follower_match:
c.follower_count = int(follower_match.group(1))
# 方法2: JavaScript精确提取
if c.follower_count == 0:
follower_count = await page.evaluate("""
() => {
const elements = document.querySelectorAll('*');
for (const el of elements) {
const text = el.innerText?.trim();
if (text === '粉丝' || text === '粉丝数') {
const parent = el.parentElement;
if (parent && !parent.innerText.includes('获赞')) {
const match = parent.innerText.match(/([\\d,\\.]+)\\s*[万wKk]?/);
if (match) return match[0];
}
}
}
return '';
}
""")
if follower_count:
c.follower_count = self.extract_num(follower_count)
# 修正名字
if c.creator_name.startswith("User-"):
name_match = re.search(r'"nickname":"([^"]+)"', author_html)
if name_match:
c.creator_name = self.fix_unicode(name_match.group(1))
except Exception as e:
self.log(f"[WARN] 粉丝数获取失败: {str(e)[:40]}")
return c
except Exception as e:
self.log(f"[ERR] {str(e)[:50]}")
return None
def export(self):
if not self.all_results:
self.log("\n[RESULT] 无数据")
return
self.log(f"\n{'='*50}")
self.log(f"[RESULT] 共 {len(self.all_results)} 条")
self.log(f"{'='*50}")
# 按粉丝量排序
sorted_results = sorted(self.all_results, key=lambda x: x.follower_count, reverse=True)
data = [r.to_dict() for r in sorted_results]
for r in sorted_results[:10]: # 只显示前10
safe_name = r.creator_name.encode('ascii', 'ignore').decode('ascii') if r.creator_name else "Unknown"
self.log(f" {safe_name} | 粉丝:{r.follower_count} | 关键词:{r.matched_keyword}")
if len(sorted_results) > 10:
self.log(f" ... 还有 {len(sorted_results)-10} 条")
ts = self.session_id
if self.output_format in ["excel", "both"]:
excel = self.output_dir / f"creators_{ts}.xlsx"
pd.DataFrame(data).to_excel(excel, index=False, engine='openpyxl')
self.log(f"\n[OK] Excel: {excel}")
if self.output_format in ["json", "both"]:
json_file = self.output_dir / f"creators_{ts}.json"
with open(json_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
self.log(f"[OK] JSON: {json_file}")
if __name__ == "__main__":
bot = XiaoHongShuBot()
asyncio.run(bot.run())
FILE:docs/README.md
# 小红书AI创作者识别 Skill
## 功能
自动在小红书平台搜索AI原创动画创作者，按预设规则筛选，导出结构化数据。
## 快速开始
### 1. 安装依赖
```bash
cd ~/.openclaw/workspace/skills/xiaohongshu-creator-finder
pip install -r requirements.txt
playwright install chromium
```
### 2. 首次运行（需要登录）
**方式1：使用OpenClaw命令**
```
/小红书创作者识别
```
**方式2：直接运行Python脚本**
```bash
# 基础搜索
python skill.py
# 自定义参数
python skill.py --keywords "AI动画,AI短剧" --min-likes 100 --min-followers 1000
# 使用配置文件
python skill.py --config config.example.json
```
### 3. 登录
首次运行会弹出浏览器窗口，请用手机小红书APP扫描二维码登录。
登录成功后，cookie会自动保存，下次运行无需再次登录。
## 参数说明
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| keywords | array | ["AI动画","AI短剧"] | 搜索关键词 |
| min_likes | number | 50 | 最小点赞数 |
| min_comments | number | 5 | 最小评论数 |
| max_days_ago | number | 30 | 最大发布时间（天前） |
| min_followers | number | 500 | 最小粉丝量 |
| max_results | number | 30 | 每个关键词最大采集数 |
| output_format | string | excel | 输出格式 |
| headless | boolean | false | 无界面模式 |
## 命令行用法
```bash
# 基础用法
python skill.py
# 指定关键词
python skill.py -k "AI动画,AI短剧,AI视频"
# 提高筛选门槛
python skill.py --min-likes 200 --min-followers 2000 --max-days 15
# 无界面模式（仅已登录后可用）
python skill.py --headless
# 导出为JSON
python skill.py -o json
```
## 输出文件
运行完成后，在 `output/` 目录生成：
```
xiaohongshu_creators_20240317_143022.xlsx # Excel表格
xiaohongshu_creators_20240317_143022.json # JSON数据
```
## 表格字段
| 字段 | 说明 |
|------|------|
| 平台 | 小红书 |
| 创作者ID | 用户ID |
| 创作者名称 | 昵称 |
| 主页链接 | 小红书主页URL |
| 粉丝量 | 粉丝数 |
| 视频标题 | 匹配视频标题 |
| 视频链接 | 视频详情页 |
| 点赞数 | 视频点赞 |
| 评论数 | 视频评论 |
| 发布日期 | 视频发布时间 |
| 匹配关键词 | 搜索匹配的关键词 |
| 采集时间 | 数据抓取时间 |
## 注意事项
1. **反爬风险**：小红书有反爬机制，已内置随机延迟，建议不要频繁运行
2. **登录状态**：Cookie可能过期，如遇问题请删除 `cookies.json` 重新登录
3. **页面改版**：如解析失败，可能需要更新CSS选择器
4. **合规使用**：请遵守平台规则，仅供学习和研究使用
## 故障排查
### 无法登录
- 检查网络是否能访问 xiaohongshu.com
- 删除 `cookies.json` 重新扫码
- 确保手机小红书APP已登录
### 找不到视频
- 检查关键词是否有效
- 尝试降低筛选条件
- 页面结构可能变更，需要更新选择器
### 数据为空
- 筛选条件可能过于严格
- 该关键词下可能无符合条件的内容
- 检查登录状态是否正常
FILE:docs/requirements.txt
playwright>=1.40.0
pandas>=2.0.0
openpyxl>=3.1.0
FILE:docs/SKILL.md
# 小红书AI创作者识别 Skill
## 名称
xiaohongshu-creator-finder
## 描述
在小红书平台按预设规则自动搜索并识别AI原创动画创作者。支持多关键词搜索、智能筛选、数据导出到Excel/JSON。
## 触发方式
### 方式1：命令触发
```
/小红书创作者识别
/找AI创作者
/xhs-finder
```
### 方式2：自然语言
- "帮我找小红书上的AI动画创作者"
- "搜索小红书的AI短剧创作者"
- "识别小红书上做AI视频的博主"
## 项目结构
```
xiaohongshu-creator-finder/
├── skill.py # 主入口文件
├── skill.json # Skill配置
├── src/
│ └── xhs_creator_finder.py # 核心程序
├── config/
│ ├── settings.json # 运行参数配置 ⭐修改这里
│ └── cookies.json # 登录Cookie（自动生成）
├── output/ # 输出结果 ⭐结果在这里
│ ├── creators_YYYYMMDD_HHMMSS.xlsx
│ └── creators_YYYYMMDD_HHMMSS.json
├── docs/
│ └── 使用说明书.md # 详细文档
└── logs/ # 运行日志
```
## 配置文件说明
编辑 `config/settings.json` 自定义搜索规则：
```json
{
"keywords": ["AI动画", "AI短剧", "AI视频"],
"max_results": 15,
"min_followers": 1000,
"min_comments": 20,
"output_format": "excel"
}
```
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `keywords` | 搜索关键词列表 | ["AI动画"] |
| `max_results` | 每个关键词搜索多少个视频 | 15 |
| `min_followers` | 最小粉丝量（过滤条件） | 1000 |
| `min_comments` | 最小评论数（过滤条件） | 20 |
| `output_format` | 输出格式 | excel |
| `chrome_path` | Chrome浏览器路径 | 自动检测 |
## 输出结果
### 文件位置
```
output/creators_YYYYMMDD_HHMMSS.xlsx
output/creators_YYYYMMDD_HHMMSS.json
```
### 数据字段
- 平台、创作者ID、创作者名称
- 主页链接、粉丝量
- 视频链接、点赞数、评论数
- 匹配关键词、采集时间
## 首次使用
1. 触发命令 `/小红书创作者识别`
2. 弹出Chrome窗口，**扫码登录**小红书
3. Cookie自动保存，后续无需登录
4. 等待运行完成，查看 `output/` 目录结果
## 注意事项
1. **运行时间**: 1个关键词约3-4分钟（受OpenClaw 5分钟超时限制）
2. **频率控制**: 建议间隔5-10分钟再运行
3. **Cookie过期**: 如遇登录问题，删除 `config/cookies.json` 重新登录
4. **筛选说明**:
- `[FILTER]` = 粉丝/评论数不达标
- `[SKIP]` = 非视频内容
- `[FAIL]` = 解析失败
## 依赖
- Python 3.8+
- Playwright
- pandas
- openpyxl
```bash
pip install playwright pandas openpyxl
playwright install chromium
```
## 详细文档
查看 `docs/使用说明书.md` 获取完整使用指南。
FILE:docs/STATUS.md
# 小红书AI创作者识别 - 当前状态
## 问题诊断
**已完成:**
- ✅ Skill 框架搭建完成
- ✅ 本地 Chrome 浏览器调用成功
- ✅ 登录流程正常
- ✅ Cookie 持久化工作
**待解决:**
- ⚠️ 页面解析 - 小红书的反爬机制和动态加载导致选择器失效
## 原因分析
小红书网页版的特点：
1. **动态渲染** - 内容通过JavaScript动态加载
2. **反爬机制** - CSS类名随机化，传统选择器难以定位
3. **需要交互** - 需要滚动、等待才能加载完整内容
## 推荐解决方案
### 方案A：使用官方API（推荐）
申请小红书开放平台API，通过官方接口获取数据。
优点：稳定、不会被封
缺点：需要企业资质申请
### 方案B：使用第三方数据服务
使用如考古加、蝉妈妈等第三方数据平台的API。
优点：即开即用
缺点：需要付费
### 方案C：增强版Playwright爬虫
使用更智能的页面解析策略：
1. 截图 + OCR识别
2. 模拟真实用户操作
3. 使用视觉定位而非CSS选择器
### 方案D：移动端抓包
通过抓包小红书APP的API请求，直接调用其内部接口。
优点：数据完整
缺点：技术门槛高，可能违反协议
## 当前Skill代码位置
```
~/.openclaw/workspace/skills/xiaohongshu-creator-finder/
├── skill.py # 主程序
├── diagnose.py # 诊断工具
└── README.md # 文档
```
## 建议下一步
由于网页反爬复杂，建议转向：
1. **飞书文档API集成** - 将数据导出到飞书
2. **定时任务** - 设置每周自动运行
3. **多平台版本** - 先做抖音/快手（相对容易）
需要我帮你实现哪个方向？
FILE:docs/使用说明书.md
# 小红书AI创作者识别 - 使用说明书
## 📌 项目简介
自动搜索小红书平台上的AI创作者（AI动画、AI短剧、AI视频等），根据预设规则筛选符合条件的创作者，并导出到Excel/JSON文件。
---
## 🚀 快速开始
### 1. 触发方式
在OpenClaw中输入以下任一命令：
```
/小红书创作者识别
/找AI创作者
/xhs-finder
```
或自然语言：
- "帮我找小红书上的AI动画创作者"
- "搜索小红书的AI短剧创作者"
### 2. 运行流程
```
触发命令 → 打开Chrome → 登录小红书 → 搜索关键词 →
筛选创作者 → 导出文件 → 展示结果
```
---
## 💾 断点续跑功能（重要）
由于OpenClaw有**5分钟超时限制**，本工具支持**断点续跑**功能：
### 工作原理
1. **每处理完一个关键词，自动保存进度** 到 `output/checkpoint.json`
2. **超时中断后，下次运行自动从上次进度继续**
3. **所有关键词处理完成后，自动导出最终结果**
### 使用示例
假设你要处理10个关键词，每次只能处理2个：
**第一次运行：**
```
[CHECKPOINT] 已加载进度: 0/10 个关键词完成
[CHECKPOINT] 待处理关键词: ['AI动画', 'AI短剧', 'AI视频', ...]
...运行中...
[CHECKPOINT] 已保存进度: 2/10 ← 超时中断
```
**第二次运行（同一会话ID）：**
```
[CHECKPOINT] 已加载进度: 2/10 个关键词完成
[CHECKPOINT] 待处理关键词: ['AI视频', 'AI绘画', ...] ← 自动跳过已处理的
...继续运行...
[CHECKPOINT] 已保存进度: 4/10 ← 再次超时
```
**直到全部完成：**
```
[CHECKPOINT] 已加载进度: 10/10 个关键词完成
[RESULT] 共 XX 条
[CHECKPOINT] 已清除断点文件
```
### 文件位置
- **断点文件:** `output/checkpoint.json`
- **结果文件:** `output/creators_YYYYMMDD_HHMMSS.xlsx`
### 注意事项
1. **同一会话ID** - 同一次任务的所有运行使用相同的session_id，结果会合并
2. **断点自动清理** - 任务完成后会自动删除checkpoint.json
3. **重复数据自动去重** - 基于创作者ID自动去重
4. **如需重新开始** - 手动删除 `output/checkpoint.json`
---
## 📁 文件目录结构
```
xiaohongshu-creator-finder/
├── skill.py # 主入口文件（触发点）
├── skill.json # Skill配置文件
│
├── src/
│ └── xhs_creator_finder.py # 核心爬虫程序
│
├── config/
│ ├── config.example.json # 配置模板
│ ├── cookies.json # 登录Cookie（自动保存）
│ └── settings.json # 运行参数配置
│
├── output/ # 输出结果目录
│ ├── creators_YYYYMMDD_HHMMSS.xlsx # Excel结果
│ ├── creators_YYYYMMDD_HHMMSS.json # JSON结果
│ └── checkpoint.json # 断点续跑进度文件（自动生成）
│
├── docs/
│ ├── 使用说明书.md # 本文件
│ ├── SKILL.md # Skill定义文档
│ ├── README.md # 项目说明
│ └── requirements.txt # Python依赖
│
└── logs/ # 运行日志
└── run_YYYYMMDD.log
```
---
## ⚙️ 配置说明
### 修改搜索规则
编辑 `config/settings.json`：
```json
{
"keywords": ["AI动画", "AI短剧", "AI视频"],
"max_results": 15,
"min_followers": 1000,
"min_comments": 20,
"output_format": "excel"
}
```
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `keywords` | 搜索关键词列表 | ["AI动画"] |
| `max_results` | 每个关键词搜索多少个视频 | 15 |
| `min_followers` | 最小粉丝量（少于这个数会过滤） | 1000 |
| `min_comments` | 最小评论数（少于这个数会过滤） | 20 |
| `output_format` | 输出格式：excel/json/feishu | excel |
### 修改关键词示例
搜索AI绘画创作者：
```json
{
"keywords": ["AI绘画", "AI插画", "AI绘图"]
}
```
放宽筛选条件：
```json
{
"min_followers": 500,
"min_comments": 10
}
```
---
## 📊 输出结果
### 文件位置
```
output/creators_YYYYMMDD_HHMMSS.xlsx
output/creators_YYYYMMDD_HHMMSS.json
```
### 数据字段
| 字段 | 说明 | 示例 |
|------|------|------|
| 平台 | 固定为"小红书" | 小红书 |
| 创作者ID | 用户唯一ID | 62d12f400000000002002b3d |
| 创作者名称 | 用户昵称 | 柔树特效-AIGC动画-后期 |
| 主页链接 | 用户主页URL | https://www.xiaohongshu.com/user/profile/... |
| 粉丝量 | 粉丝数量 | 108000 |
| 视频链接 | 匹配的视频URL | https://www.xiaohongshu.com/explore/... |
| 点赞数 | 视频点赞数 | 1234 |
| 评论数 | 视频评论数 | 456 |
| 关键词 | 匹配的关键词 | AI动画 |
| 采集时间 | 数据采集时间 | 2026-03-18 11:04:17 |
---
## 🔧 常见问题
### Q1: 第一次使用需要登录吗？
**A:** 是的。首次运行会弹出Chrome窗口，需要扫码登录小红书。Cookie会自动保存，后续运行无需再次登录。
### Q2: 运行时间多久？超时怎么办？
**A:**
- 1个关键词 + 15个视频 ≈ 3-4分钟
- 3个关键词 + 30个视频 ≈ 10-15分钟
**注意:** OpenClaw单次命令有5分钟超时限制。
**解决方案 - 断点续跑：**
1. 本工具支持**断点续跑**，每处理完一个关键词自动保存进度
2. 超时中断后，**再次运行同一命令**，会自动从上次进度继续
3. 重复运行直到所有关键词处理完成，最终导出合并结果
**示例:**
```
第一次运行: 处理完2个关键词后超时
第二次运行: 自动跳过已处理的2个，继续处理第3、4个
第三次运行: 继续处理第5、6个...
直到全部完成，自动导出结果
```
### Q3: 为什么会"FILTER"或"SKIP"？
**A:**
- `[FILTER]` - 不符合筛选条件（粉丝/评论不够）
- `[SKIP]` - 非视频内容（图文笔记）
- `[FAIL]` - 页面解析失败
### Q4: 如何导出到飞书？
**A:** 修改配置：
```json
{
"output_format": "feishu",
"feishu_sheet_url": "https://..." // 飞书表格链接
}
```
### Q5: Cookie过期怎么办？
**A:** 删除 `config/cookies.json` 文件，重新运行并扫码登录。
### Q6: checkpoint.json是什么？
**A:** 这是**断点续跑**的进度文件，包含：
- 已完成的关键词列表
- 已采集的创作者数据
- 会话ID和时间戳
**什么时候删除？**
- 任务**正常完成**后会自动删除
- 如需**重新开始**，手动删除此文件
**文件位置:** `output/checkpoint.json`
---
## 📝 高级用法
### 批量运行多个关键词
编辑 `config/settings.json`：
```json
{
"keywords": [
"AI动画",
"AI短剧",
"AI视频",
"AI绘画",
"AIGC"
],
"max_results": 10
}
```
**提示:** 关键词越多，运行时间越长。建议分批执行。
### 调整筛选严格度
**严格模式**（只找头部创作者）：
```json
{
"min_followers": 10000,
"min_comments": 100
}
```
**宽松模式**（找新兴创作者）：
```json
{
"min_followers": 100,
"min_comments": 5
}
```
---
## ⚠️ 注意事项
1. **遵守平台规则** - 请合理使用，不要频繁抓取
2. **控制频率** - 建议每次间隔5-10分钟
3. **数据准确性** - 由于网页动态加载，数据可能存在延迟
4. **网络环境** - 需要稳定的网络连接
---
## 🛠️ 技术信息
- **开发语言:** Python 3.8+
- **浏览器:** Chrome/Chromium
- **框架:** Playwright
- **数据导出:** pandas + openpyxl
### 依赖安装
```bash
pip install playwright pandas openpyxl
playwright install chromium
```
---
## 📧 问题反馈
如有问题，请检查：
1. Chrome浏览器是否安装
2. Python依赖是否完整
3. 网络连接是否正常
4. 小红书账号是否正常
---
*最后更新: 2026-03-18*
FILE:config/config.example.json
{
"keywords": ["AI动画", "AI短剧", "AI视频", "AI绘画"],
"min_likes": 100,
"min_comments": 10,
"max_days_ago": 30,
"min_followers": 1000,
"max_results": 50,
"output_format": "excel",
"headless": false
}
FILE:config/cookies.json
[
{
"name": "acw_tc",
"value": "0a00d76417737369506113574e7227426a2755f9e111eee9324d15639d2565",
"domain": "www.xiaohongshu.com",
"path": "/",
"expires": 1773738750.694486,
"httpOnly": true,
"secure": false,
"sameSite": "Lax"
},
{
"name": "abRequestId",
"value": "05c40e2d-cc5b-5989-a5bc-169960027915",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805272950.694653,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "webBuild",
"value": "6.1.0",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": -1,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "xsecappid",
"value": "xhs-pc-web",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805273657,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "a1",
"value": "19cfaf5fe2bkyljhif99iauaoswgqizazhxipmivp50000152666",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805272951,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "webId",
"value": "a3f0763fa1650619470a1c4ccb01c971",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805272951,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "acw_tc",
"value": "0a4ae09717737369516868160e8bea0b70d7e9244e35b4f9ff95438b29932c",
"domain": "edith.xiaohongshu.com",
"path": "/",
"expires": 1773738751.669109,
"httpOnly": true,
"secure": false,
"sameSite": "Lax"
},
{
"name": "acw_tc",
"value": "0a4a77a817737369523627464ec35bea9ab9e342f4972c57abcfea8ec322cf",
"domain": "so.xiaohongshu.com",
"path": "/",
"expires": 1773738752.331931,
"httpOnly": true,
"secure": false,
"sameSite": "Lax"
},
{
"name": "gid",
"value": "yjSi0iKyjju2yjSi0i2idx63JDTv1Vx3ijj3ChU070lME328kA3u0U888y2JKKK8SWD22dWq",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1808297661.092483,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "web_session",
"value": "0400698e5acc1556c4786b5c8c3b4bb4b7bc00",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805272962.554918,
"httpOnly": true,
"secure": true,
"sameSite": "Lax"
},
{
"name": "id_token",
"value": "VjEAAEU6d/ZUqfkJueM+W4Lr6aM4oxNDoWDlRGsi4vqVkdOJtSA0dYgV29PFcfZ9w2Yc2ZOdJg8u4vAWS8Ht8BO+JcrLfRzj2UxChhn7yPJbyxsoC2s08hk/2Dp3j95+NTwheGu9",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805272962.555022,
"httpOnly": true,
"secure": true,
"sameSite": "Lax"
},
{
"name": "websectiga",
"value": "6169c1e84f393779a5f7de7303038f3b47a78e47be716e7bec57ccce17d45f99",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1773996807,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "sec_poison_id",
"value": "ed593592-a2da-465c-be27-0d187cc5a21d",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1773738212,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "loadts",
"value": "1773737657776",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1805273657,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
},
{
"name": "ets",
"value": "1773737661351",
"domain": ".xiaohongshu.com",
"path": "/",
"expires": 1776329661.355011,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
}
]
FILE:config/settings.json
{
"keywords": ["AI动画", "AI短剧", "AI视频"],
"max_results": 15,
"min_followers": 1000,
"min_comments": 20,
"output_format": "excel",
"headless": false,
"delay_between_requests": 2,
"chrome_path": "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe"
}
智能简历项目总结助手。当用户说"总结项目"、"归纳项目"、"整理项目"时，帮助用户从多渠道获取项目信息，按岗位方向（产品/运营/技术/设计/数据等）归纳整理为高质量互联网简历，提升面试通过率。
---
name: resume-project-summarizer
description: 智能简历项目总结助手。当用户说"总结项目"、"归纳项目"、"整理项目"时，帮助用户从多渠道获取项目信息，按岗位方向（产品/运营/技术/设计/数据等）归纳整理为高质量互联网简历，提升面试通过率。
triggers:
- 总结项目
- 归纳项目
- 整理项目
- 写简历
- 优化简历
- 项目经历
---
# 简历项目总结助手
将零散的项目经历整理为结构化的互联网方向简历，支持**产品、运营、技术、设计、数据**等多个方向。
## 支持的岗位方向
| 大类 | 具体岗位 |
|:---|:---|
| **产品类** | 产品经理、策略产品、商业化产品、数据产品 |
| **运营类** | 产品运营、用户运营、内容运营、活动运营、社群运营、新媒体运营 |
| **技术类** | Java后端、Go后端、前端开发、算法工程师、测试开发、运维工程师 |
| **设计类** | UI设计、UX设计、视觉设计、交互设计 |
| **数据类** | 数据分析、数据挖掘、商业分析 |
| **其他** | 市场营销、销售、HR、财务等（通用模板） |
## 工作流程
### Step 1: 信息收集
主动询问用户提供项目经历信息：
> 我可以帮你整理项目经历，生成针对互联网岗位的简历。请先提供你的项目信息：
>
> **方式1 - 粘贴链接**：飞书文档、企业微信文档、语雀等在线文档链接
> **方式2 - 上传文件**：PDF、Word、图片格式的简历或项目总结
> **方式3 - 文字描述**：直接输入项目经历的文字描述
>
> 你方便用哪种方式？
**支持的输入源：**
- 飞书文档（/docx/xxx 或 /wiki/xxx）
- 企业微信文档（需要分享为外部可访问链接）
- 语雀文档
- PDF / Word / TXT 文件
- 图片（OCR识别）
- 直接文字输入
### Step 2: 信息解析与提取
从用户提供的内容中提取以下结构化字段：
```yaml
项目信息结构:
基本信息:
- 公司名称/项目类型
- 项目时间（起止年月）
- 岗位/角色（产品实习生、运营实习生、Java开发实习生等）
- 所在部门/团队
项目背景:
- 业务背景（为什么要做这个项目）
- 核心目标（要解决什么问题）
- 目标用户群体
个人职责:
- 负责的具体模块/功能
- 使用的技术栈/工具（Java、Python、Figma、SQL等）
- 协作角色（与哪些团队合作）
- 独立负责还是参与支持
项目成果:
- 量化数据（用户数、转化率、接口QPS、页面性能提升等）
- 定性成果（用户好评、获得奖项、领导认可）
- 对公司/业务的实际影响
能力标签:
- 硬技能（技术栈、工具、方法论等）
- 软技能（跨部门协作、项目管理、沟通表达等）
```
**信息提取规则：**
- 自动识别时间格式（2024.06 / 2024年6月 / June 2024）
- 自动识别数字和单位（143w+用户、接口QPS提升50%、页面加载优化30%）
- 自动提取技能关键词（技术栈、设计工具、分析方法等）
### Step 3: 杂乱信息整理（关键步骤）
用户提供的原始信息往往比较杂乱，需要进行智能整理：
#### 3.1 信息去噪
**去除无关信息：**
- 删除日常琐事（开会、日报、周报等）
- 删除与岗位无关的内容（如投技术岗时删除纯行政事务）
- 删除过于笼统的描述（"参与了一些项目"）
- 删除重复信息
**示例：**
```
❌ 原始信息：
"每天参加早会，写日报周报，帮领导订过咖啡，参与了部门团建，
做过一个用户增长项目，还有一个数据分析的工作，也写过PRD文档"
✅ 整理后：
- 用户增长项目（需进一步询问细节）
- 数据分析工作（需进一步询问细节）
- PRD文档撰写（需进一步询问具体项目）
```
#### 3.2 项目识别与拆分
**识别核心项目：**
- 从杂乱描述中识别出**独立的项目单元**
- 将混在一起的多段经历**拆分**成独立条目
- 识别项目的**核心内容**和**边缘内容**
**拆分示例：**
```
❌ 原始信息：
"我在字节实习了6个月，做了很多事。前两个月做抖音的内容推荐策略，
后来做增长活动，还帮运营做了一些数据分析，最后一个月做AB测试平台"
✅ 拆分后：
项目1：抖音内容推荐策略优化（前2个月）
项目2：抖音增长活动策划与执行（中间2个月）
项目3：数据分析支持（贯穿或特定时期）
项目4：AB测试平台建设（最后1个月）
```
#### 3.3 时间线梳理
**理清时间顺序：**
- 识别所有时间信息，建立**时间轴**
- 发现**时间冲突**或**空白期**，标记询问
- 按时间倒序排列（最近→最早）
**时间整理示例：**
```
❌ 原始信息：
"做了一个春节活动，然后之前做过一个国庆活动，还有一个日常的需求分析"
✅ 整理后：
- 2024.01-2024.02：春节活动（最近）
- 2023.09-2023.10：国庆活动
- 2023.10-2024.01：日常需求分析（需确认具体时间）
```
#### 3.4 重要性评估
**判断信息重要程度：**
| 优先级 | 特征 | 处理方式 |
|:---|:---|:---|
| **P0 - 核心项目** | 有数据成果、独立负责、与岗位高度相关 | 重点展开，放在简历最前面 |
| **P1 - 重要项目** | 有参与、有学习价值、能体现能力 | 正常描述，2-3个bullet point |
| **P2 - 辅助信息** | 参与支持、技能相关但成果不突出 | 简写或合并到其他项目 |
| **P3 - 可以删除** | 日常事务、与岗位无关、过于笼统 | 直接删除 |
**重要性评估示例：**
```
原始信息：
1. 独立负责用户增长策略，DAU提升30% → 【P0】核心项目
2. 协助做了一个数据分析报告 → 【P1】重要项目（可展开）
3. 参加了一些用户访谈 → 【P2】辅助信息（可合并到项目1）
4. 帮运营整理了Excel表格 → 【P3】删除
5. 每周写周报 → 【P3】删除
```
#### 3.5 关联性合并
**识别项目间的关联：**
- **同一业务的多个环节**：合并为一个完整项目
- **同一技术的多个应用**：合并展示技术深度
- **同一时期的相关工作**：按模块合并
**合并示例：**
```
原始信息：
- 做了搜索功能的优化
- 做了推荐功能的优化
- 做了一周数据分析
✅ 关联合并后：
- 搜索与推荐策略优化（核心项目）
- 搜索优化：Query理解、结果排序
- 推荐优化：召回策略、排序模型
- 数据支持：用户行为分析、效果评估
```
#### 3.6 向用户确认整理结果
整理完成后，向用户展示整理结果并确认：
> 我已从你提供的信息中整理出以下核心内容：
>
> **识别到的公司/组织**：
> 1. 字节跳动 - 产品实习生（2024.01-2024.06）
> 2. 腾讯 - 运营实习生（2023.07-2023.12）
>
> **从字节经历中拆分出的核心项目**：
> - [P0] 抖音搜索策略优化（独立负责，DAU提升数据）
> - [P1] 春节增长活动（主导，用户增长数据）
> - [P1] 数据分析支持（协助，输出分析报告）
> - [P3] 日常事务（已过滤：周报、会议记录等）
>
> **请确认：**
> 1. 以上拆分是否正确？有没有遗漏或合并错误？
> 2. 【P0】项目是否有更多细节补充？
> 3. 【P3】被过滤的内容中，是否有需要保留的？
### Step 4: 项目关联合并
按以下规则处理多个项目：
**合并策略：**
1. **同一公司合并**：同一公司的多段经历合并为一段连续实习
2. **相似项目去重**：内容高度相似的项目只保留最完整版本
3. **时间排序**：按时间倒序排列（最近的经历放前面）
**去重规则：**
- 项目描述相似度 > 80% 视为重复
- 保留数据更完整、描述更详细的版本
- 标记存疑项目询问用户确认
### Step 5: 岗位方向确认
询问用户目标岗位方向：
> 已提取到 【X】 段项目经历。接下来请确认你的简历方向：
>
> **目标岗位方向**（可多选）：
>
> **【产品类】**
> - [ ] 产品经理（需求分析、产品设计）
> - [ ] 策略产品（推荐/搜索策略、算法应用）
> - [ ] 商业化产品（广告、变现、收入优化）
> - [ ] 数据产品（数据平台、指标体系）
>
> **【运营类】**
> - [ ] 产品运营（活动策划、增长策略）
> - [ ] 用户运营（用户分层、生命周期管理）
> - [ ] 内容运营（创作者运营、内容策略）
> - [ ] 新媒体运营（短视频、公众号、社群）
>
> **【技术类】**
> - [ ] Java后端开发（Spring、微服务、高并发）
> - [ ] Go后端开发（云原生、中间件、分布式）
> - [ ] 前端开发（React/Vue、性能优化、工程化）
> - [ ] 算法工程师（机器学习、NLP、推荐算法）
> - [ ] 测试开发（自动化测试、质量保障）
> - [ ] 运维/SRE（DevOps、云原生、稳定性）
>
> **【设计类】**
> - [ ] UI设计（界面设计、设计系统）
> - [ ] UX/交互设计（用户研究、交互流程）
> - [ ] 视觉设计（品牌、运营视觉、插画）
>
> **【数据类】**
> - [ ] 数据分析（业务分析、指标体系、A/B测试）
> - [ ] 数据挖掘（机器学习、特征工程、模型应用）
> - [ ] 商业分析（战略分析、商业洞察）
>
> **【其他】**
> - [ ] 市场营销（品牌、推广、公关）
> - [ ] 销售/BD（客户拓展、商务合作）
> - [ ] 职能类（HR、财务、行政）
> - [ ] 其他：_______
>
> **求职类型**：
> - [ ] 校招（202X届）
> - [ ] 暑期实习
> - [ ] 日常实习
> - [ ] 社招
>
> **目标公司类型**：
> - [ ] 互联网大厂（字节、腾讯、阿里等）
> - [ ] 中型互联网公司
> - [ ] 创业公司
> - [ ] 外企
> - [ ] 传统行业数字化
### Step 6: 信息补全确认
针对提取的信息缺失或不明确之处，逐一询问用户。
**不同岗位的补全重点：**
**技术类岗位重点询问：**
- 使用的技术栈？（语言、框架、中间件、数据库）
- 系统的QPS/TPS/并发量？
- 解决了什么技术难题？性能优化了多少？
- 代码量/接口数/模块复杂度？
**设计类岗位重点询问：**
- 设计工具？（Figma、Sketch、PS、AI等）
- 设计成果？（页面数、组件库、设计系统）
- 设计验证方式？（用户测试、AB测试、数据反馈）
- 设计影响力？（用户满意度提升、转化率提升）
**运营类岗位重点询问：**
- 负责的数据指标？（DAU、留存、GMV、内容量）
- 具体的运营策略和执行动作？
- 活动效果数据？（参与人数、转化率、ROI）
- 用户分层和精细化运营方法？
**产品类岗位重点询问：**
- 需求分析和方案设计过程？
- 跨部门协作和项目推动？
- 数据验证和效果评估？
### Step 6: 简历生成
根据确认的岗位方向，生成针对性简历。
**各岗位侧重点映射：**
| 岗位方向 | 简历侧重点 | STAR描述重点 | 关键词 |
|:---|:---|:---|:---|
| **产品经理** | 需求洞察、方案设计、推动落地、效果验证 | 用户调研→PRD设计→跨团队协作→上线数据 | 用户研究、产品设计、A/B测试、数据驱动、跨部门推动 |
| **策略产品** | 算法应用、策略设计、效果评估、模型优化 | 业务目标→策略拆解→模型应用→效果迭代 | 推荐策略、搜索策略、Prompt工程、模型评估、特征工程 |
| **商业化产品** | 变现模式、广告策略、收入优化、ROI提升 | 收入目标→策略设计→AB测试→收入提升 | ARPU、CPM、ROI、广告主体验、变现效率 |
| **产品运营** | 活动策划、增长策略、数据分析、用户分层 | 活动目标→策略设计→资源整合→数据复盘 | 用户增长、留存提升、GMV、DAU、转化率、裂变 |
| **用户运营** | 用户分层、生命周期、精细化运营、留存提升 | 用户分析→分层策略→触达手段→效果评估 | LTV、留存率、活跃度、用户分层、触达策略 |
| **内容运营** | 内容策略、创作者运营、生态建设、爆款打造 | 内容规划→创作者激励→分发策略→生态健康 | UGC、PGC、创作者、内容质量、分发效率 |
| **Java后端** | 高并发、微服务、系统架构、性能优化 | 业务需求→技术方案→编码实现→性能调优 | Spring、MySQL、Redis、Kafka、微服务、分布式 |
| **Go后端** | 云原生、中间件、高可用、高性能 | 业务场景→架构设计→编码实现→性能优化 | Golang、Gin、GORM、Docker、K8s、etcd |
| **前端开发** | 工程化、性能优化、用户体验、组件化 | 需求分析→组件设计→编码实现→性能优化 | React、Vue、Webpack、Vite、TypeScript、工程化 |
| **算法工程师** | 模型设计、特征工程、效果优化、工程落地 | 业务问题→模型选型→特征工程→效果评估 | 机器学习、深度学习、NLP、推荐算法、特征工程 |
| **测试开发** | 质量保障、自动化测试、效能提升、稳定性 | 质量风险→测试方案→自动化建设→效能提升 | 自动化测试、接口测试、性能测试、CI/CD、质量门禁 |
| **UI设计** | 视觉设计、设计系统、组件规范、品牌一致性 | 需求理解→视觉设计→设计规范→组件沉淀 | Figma、Sketch、设计系统、组件库、视觉规范 |
| **UX设计** | 用户研究、交互流程、可用性、体验优化 | 用户研究→信息架构→交互设计→可用性测试 | 用户访谈、 personas、 journey map、原型设计、可用性测试 |
| **数据分析** | 指标体系、归因分析、模型应用、可视化 | 业务问题→数据拆解→分析建模→结论输出 | SQL、Python、指标体系、归因分析、漏斗分析、用户画像 |
**简历生成规则：**
1. **格式规范**
- 一页纸原则（校招/实习）
- 倒序时间线
- bullet point 不超过3行
2. **STAR法则优化**
- Situation（背景）: 1句话，说明项目目标和挑战
- Task（任务）: 1句话，说明你的职责
- Action（行动）: 2-3点，具体做了什么
- Result（结果）: 量化数据前置，突出成果
3. **数据前置原则**
```
❌ 原写法：
负责用户增长策略设计，通过活动策划提升了用户量。
✅ 优化写法：
用户增长XX万（+XX%）| 设计并落地增长策略，策划XX活动，通过裂变+投放组合实现用户规模突破
```
4. **技术类简历特殊规则**
- 技术栈明确写出（Java/Spring/Redis/MySQL）
- 性能数据量化（QPS从X提升到X、响应时间从Xms降到Xms）
- 突出技术难点和解决方案
- 代码质量和工程实践（单元测试、Code Review、CI/CD）
5. **设计类简历特殊规则**
- 设计成果可视化（页面数、组件数、设计系统覆盖度）
- 设计验证数据（用户满意度、转化率提升、NPS）
- 设计思考和设计过程
- 工具和方法论
6. **关键词植入**
- 根据岗位JD自动植入相关关键词
- 确保ATS（applicant tracking system）能识别
### Step 7: 增值内容生成
除了简历本身，还提供：
**A. 简历优化建议**
- 亮点总结
- 可提升点
- 针对目标岗位的具体建议
**B. 面试问题预测**
- 根据岗位类型预测10-15个面试问题
- 技术类包含算法题和系统设计题
- 设计类包含作品解读和设计思路题
**C. 投递建议**
- 匹配岗位推荐
- 建议避开的岗位
- 简历投递策略
### Step 8: 输出交付
提供多种格式选项：
> 简历已生成！请选择输出格式：
>
> **格式选项**：
> - [ ] Markdown格式（适合复制到飞书/语雀）
> - [ ] 纯文本格式（适合粘贴到招聘网站）
> - [ ] Word文档（.docx，可直接编辑美化）
>
> **附加内容**：
> - [✓] 简历优化建议
> - [✓] 面试问题预测（10-15题）
> - [ ] 生成其他岗位版本
## 隐私与安全声明
**数据处理原则：**
- 用户提供的项目信息仅用于生成简历，不长期存储
- 处理完成后可主动删除临时文件
- 敏感信息（手机号、具体薪资等）可脱敏处理
- 飞书/企微文档链接需要用户确认授权范围
## 版本记录
- v1.0: 基础版本，支持产品类岗位
- v2.0: 扩展支持运营、技术、设计、数据等多类岗位
- v2.1: 增加技术类性能指标提取、设计类作品展示建议
FILE:_meta.json
{
"name": "resume-project-summarizer",
"version": "2.1.0",
"description": "智能简历项目总结助手，支持产品、运营、技术、设计、数据等多类岗位。智能处理杂乱信息，自动去噪、拆分、关联、排序，生成高质量互联网简历",
"author": "HeyWeekly",
"triggers": ["总结项目", "归纳项目", "整理项目", "写简历", "优化简历", "项目经历"],
"input_methods": ["飞书链接", "企微链接", "PDF", "Word", "图片", "文字输入"],
"output_formats": ["Markdown", "Word", "纯文本"],
"job_categories": {
"产品类": ["产品经理", "策略产品", "商业化产品", "数据产品"],
"运营类": ["产品运营", "用户运营", "内容运营", "活动运营", "社群运营", "新媒体运营"],
"技术类": ["Java后端", "Go后端", "前端开发", "算法工程师", "测试开发", "运维工程师"],
"设计类": ["UI设计", "UX设计", "视觉设计", "交互设计"],
"数据类": ["数据分析", "数据挖掘", "商业分析"],
"其他": ["市场营销", "销售", "HR", "财务"]
},
"features": [
"智能信息去噪 - 自动过滤无关日常事务",
"项目识别拆分 - 从杂乱描述中提取核心项目",
"时间线梳理 - 自动排序并发现时间冲突",
"重要性评估 - P0/P1/P2/P3分级",
"关联性合并 - 智能合并相关项目",
"15+岗位方向智能匹配",
"STAR法则自动优化",
"技术类性能指标提取",
"设计类作品展示建议",
"面试问题智能预测"
],
"privacy": "用户数据仅用于生成简历，不长期存储",
"created_at": "2026-03-06",
"updated_at": "2026-03-06"
}
FILE:references/简历写作技巧.md
# 简历写作技巧大全
## 一、简历核心原则
### 一页纸原则
- **校招/实习**：严格一页纸
- **社招3年以内**：尽量一页纸
- **社招3年以上**：可适当增加，但不超过2页
### 倒序原则
- 最近的经历放最前面
- 每个模块内时间倒序
### 相关性原则
- 只写与目标岗位相关的经历
- 不相关的经历可以简写或不写
---
## 二、STAR法则详解
### 什么是STAR
- **S (Situation)**：背景，为什么要做这个项目
- **T (Task)**：任务，你的职责和目标
- **A (Action)**：行动，具体做了什么
- **R (Result)**：结果，取得了什么成果
### 应用示例
**❌ 原写法**
负责用户增长策略，通过活动策划提升了用户量。
**✅ STAR优化**
- **背景**：新功能上线初期用户增长缓慢，DAU仅X万
- **任务**：作为增长owner，目标3个月内DAU提升至X万
- **行动**：
- 拆解增长漏斗，识别关键转化节点
- 设计裂变激励机制，降低参与门槛
- 协调市场、技术、设计3个团队推进
- **结果**：DAU提升至X万（+X%），超预期X%
---
## 三、数据化表达技巧
### 数据前置原则
每段经历的第一句话放核心数据。
```
❌ 负责XX项目，通过XX方法，最终用户增长X万
✅ 用户增长X万（+X%）| 负责XX项目，通过XX方法实现增长突破
```
### 常用数据维度
| 岗位 | 核心数据 |
|:---|:---|
| 产品经理 | DAU、转化率、留存率、功能使用率、项目上线率 |
| 产品运营 | 新增用户、留存率、GMV、活动参与率、ROI |
| 数据分析 | 分析项目数、指标提升幅度、模型准确率 |
| 商业化 | 收入、ARPU、CPM、广告主数、变现效率 |
| 策略产品 | 点击率、转化率、策略覆盖率、实验效果 |
### 数据表达技巧
**绝对值 vs 相对值**
```
- 绝对值：用户增长10万
- 相对值：用户增长10万（+20%）
- 对比值：用户增长10万（高于大盘15%）
```
**没有精确数据时的表达**
```
- 估算：约X万、超X万、近X万
- 范围：X-X万、数十量级
- 定性：显著提升、大幅提升、用户好评
- 对比：从第X名提升至第X名、超过竞品
```
---
## 四、关键词植入技巧
### 为什么要植入关键词
1. **ATS系统**：自动筛选简历时会匹配关键词
2. **HR快速扫读**：10秒内找关键词判断匹配度
3. **面试官引导**：方便面试官问相关问题
### 各岗位关键词清单
**产品经理**
- 用户研究、需求分析、PRD、原型设计
- A/B测试、数据分析、数据驱动
- 跨部门协作、项目管理、owner意识
- 增长、留存、转化率
**产品运营**
- 用户增长、拉新、留存、促活
- 活动策划、裂变、GMV、DAU
- 用户分层、社群运营、内容运营
- 数据分析、ROI、转化率
**数据分析**
- SQL、Python、统计学、机器学习
- 指标体系、漏斗分析、归因分析
- A/B测试、假设检验、显著性
- 数据可视化、数据看板
**商业化产品**
- 广告、CPM、CPC、eCPM、ROI
- 会员、订阅、LTV、ARPU
- 变现、收入、商业化率
- 定价策略、竞价策略
**策略产品**
- 推荐策略、搜索策略、召回、排序
- 特征工程、模型评估、A/B测试
- AI、大模型、Prompt工程
- 算法、机器学习、深度学习
---
## 五、常见错误与修正
### 错误1：描述过于笼统
```
❌ 负责用户增长工作，取得了很好的效果
✅ 负责用户增长工作，3个月内DAU从X万提升至X万（+X%）
```
### 错误2：强调"参与"弱化贡献
```
❌ 参与XX项目，协助完成XX功能
✅ 负责XX模块，独立完成PRD撰写和项目推进，上线后DAU提升X%
```
### 错误3：堆砌技术名词
```
❌ 使用React、Vue、Node.js、MySQL、Redis开发系统
✅ 负责XX系统后端开发，优化数据库查询性能，响应时间降低X%
```
### 错误4：格式不统一
```
❌ 时间格式混用（2024.06 / 2024/6 / June 2024）
✅ 统一格式：2024.06 - 2024.09
```
### 错误5：错别字和语病
- 务必检查错别字
- 注意"的/地/得"用法
- 保持主谓一致
---
## 六、不同岗位的表达侧重
### 产品经理 - 突出"决策"和"推动"
```
- 决策：基于用户调研决定[XX方向]
- 设计：输出PRD、原型，设计[XX功能]
- 推动：协调[XX团队]，推动项目按期上线
- 验证：通过A/B测试验证，[XX指标]提升X%
```
### 产品运营 - 突出"策略"和"数据"
```
- 策略：设计[XX]增长策略，覆盖[XX渠道]
- 执行：落地[XX]活动，协调[XX资源]
- 数据：监控[XX指标]，根据数据优化[XX]
- 成果：新增用户X万，ROI达X
```
### 数据分析 - 突出"分析"和"洞察"
```
- 问题：发现[XX指标]异常，定位[XX问题]
- 分析：通过[XX方法]分析，发现[XX规律]
- 建议：提出[XX建议]，预期[XX效果]
- 落地：推动优化，实际[XX指标]提升X%
```
---
## 七、简历检查清单
### 格式检查
- [ ] 一页纸（校招/实习）
- [ ] 时间格式统一
- [ ] 字体字号一致
- [ ] 对齐整齐
- [ ] 无错别字
### 内容检查
- [ ] 每段经历有数据
- [ ] 动词开头
- [ ] STAR法则完整
- [ ] 关键词植入
- [ ] 成果量化
### 岗位匹配检查
- [ ] 与JD关键词匹配
- [ ] 突出相关经历
- [ ] 弱化无关经历
- [ ] 项目排序合理
### 投递前检查
- [ ] PDF格式
- [ ] 文件名规范（姓名_学校_岗位_时间）
- [ ] 文件大小合适（<5MB）
- [ ] 链接可点击（如果是在线简历）
---
## 八、简历迭代建议
### 第一次迭代：内容完整
- 把所有经历写上去
- 不纠结措辞和数据
### 第二次迭代：STAR优化
- 每个项目按STAR法则重写
- 补充数据和细节
### 第三次迭代：精简优化
- 删减不相关内容
- 压缩到一页纸
- 优化措辞
### 第四次迭代：岗位定制
- 根据具体JD调整关键词
- 调整项目排序
- 强调相关经历
### 寻求反馈
- 给学长/朋友看
- 给目标公司的前辈看
- 根据反馈迭代
---
## 九、简历投递技巧
### 投递渠道优先级
1. 内推（成功率最高）
2. 官网投递
3. 招聘网站（BOSS、智联等）
4. 邮件投递
### 邮件投递模板
```
主题：应聘XX岗位-XX大学-姓名-手机号
尊敬的面试官：
您好！我是XX大学XX专业的XX，应聘贵司XX岗位。
附件是我的简历，主要经历包括：
1. XX公司-XX岗位，负责XX，取得XX成果
2. XX公司-XX岗位，负责XX，取得XX成果
对贵司的XX业务很感兴趣，期待有机会交流。
祝好！
姓名
手机号
邮箱
```
### 跟进技巧
- 内推后1周无回复可礼貌询问进度
- 面试后24小时内发送感谢邮件
- 被拒后可询问原因以便改进
---
## 十、常见问题解答
**Q1：没有实习经历怎么写？**
- 写课程项目、社团经历、竞赛经历
- 突出项目管理和数据分析能力
- 个人账号运营（小红书/抖音/B站）
**Q2：实习时间很短怎么写？**
- 强调成果而非时间
- 突出独立负责的部分
- 如果<3个月，可合并到一段经历
**Q3：转行怎么写简历？**
- 突出 transferable skills
- 补充目标岗位的相关项目
- 解释转行动机
**Q4：GAP year怎么解释？**
- 诚实说明原因（考研/创业/休息）
- 强调期间的学习/成长
- 不要隐瞒，会被问到
**Q5：可以包装经历吗？**
- 可以优化措辞，但不要造假
- 夸大成果会被追问细节
- 诚信是底线
FILE:references/面试问题库.md
# 分岗位面试问题库
## 一、通用行为面试题
### 自我介绍类
1. 请用2-3分钟介绍一下你自己
2. 为什么选择我们公司/这个岗位？
3. 你的职业规划是什么？
### 项目深挖类（STAR法则）
4. 请详细讲一个你觉得最有成就感的项目
5. 请讲一个失败的/遗憾的项目，如果重来你会怎么做？
6. 你在这个项目中的具体角色是什么？独立负责还是参与？
### 能力考察类
7. 描述一次你和团队成员意见不一致的经历，怎么解决的？
8. 描述一次你推动跨部门协作的经历
9. 你平时怎么学习/保持对行业的敏感度？
### 认知考察类
10. 你认为自己最大的优势和劣势是什么？
11. 你对我们的产品有什么了解？有什么建议？
12. 你有什么问题想问我的？
---
## 二、产品类岗位
### 产品经理
1. 如果让你设计一个[XX功能]，你会怎么做？
2. 怎么判断一个功能要不要做？优先级怎么排？
3. 怎么做用户调研？访谈和问卷分别适合什么场景？
4. PRD需要包含哪些内容？
5. 如果上线后数据不达预期，你会怎么分析？
6. 如果开发说需求做不了，你会怎么处理？
### 策略产品
1. 推荐系统和搜索系统有什么区别？
2. 如果推荐结果太单一，怎么优化多样性？
3. 怎么评估一个策略的效果？
4. 模型效果不够好，你会怎么优化？
5. 怎么用大模型解决业务问题？
### 商业化产品
1. 你熟悉哪些商业模式？各有什么优缺点？
2. 怎么平衡用户体验和商业收入？
3. 怎么定价？有哪些定价策略？
4. LTV和CAC怎么计算？
---
## 三、运营类岗位
### 产品运营
1. 如果让你负责一个新产品的冷启动，你会怎么做？
2. 怎么设计一个裂变活动？关键要素是什么？
3. 如果某个渠道的新增下降了30%，怎么排查？
4. 怎么提升用户留存？
5. 怎么策划一个爆款活动？
### 用户运营
1. 怎么做用户分层？
2. RFM模型怎么用？
3. 不同生命周期用户怎么运营？
4. 怎么提升用户LTV？
### 内容运营
1. 怎么激励用户生产内容（UGC）？
2. 怎么判断内容质量？
3. 怎么平衡内容质量和数量？
4. 创作者运营的核心是什么？
---
## 四、技术类岗位
### Java后端开发
#### 基础知识
1. Java集合框架有哪些？ArrayList和LinkedList区别？
2. HashMap原理？线程安全吗？ConcurrentHashMap怎么实现的？
3. JVM内存模型？GC算法有哪些？
4. 多线程编程？线程池参数？死锁怎么解决？
5. synchronized和Lock区别？
#### 框架与中间件
6. Spring IOC和AOP原理？
7. Spring Boot自动配置原理？
8. MyBatis缓存机制？
9. Redis数据类型？持久化方式？缓存穿透/击穿/雪崩怎么解决？
10. Kafka为什么快？如何保证消息不丢失？
#### 数据库
11. MySQL索引原理？B+树？
12. 怎么优化慢查询？
13. 事务ACID？隔离级别？MVCC？
14. 分库分表方案？
#### 系统设计
15. 设计一个秒杀系统？
16. 设计一个短链服务？
17. 分布式事务怎么解决？
18. 高并发系统怎么设计？
#### 项目深挖
19. 你做的项目QPS多少？怎么优化的？
20. 遇到过什么技术难题？怎么解决的？
21. 代码质量怎么保证？
---
### 前端开发
#### 基础知识
1. JS原型链？this指向？闭包？
2. 事件循环？宏任务微任务？
3. 跨域问题？解决方案？
4. 浏览器渲染原理？重绘重排？
5. 防抖节流？应用场景？
#### React/Vue
6. React生命周期？Hooks原理？
7. Vue响应式原理？虚拟DOM？diff算法？
8. 状态管理？Redux/Vuex原理？
9. 组件通信方式有哪些？
10. 性能优化手段？
#### 工程化
11. Webpack原理？loader和plugin区别？
12. 代码分割怎么做？懒加载实现？
13. 前端监控怎么做？错误上报？
#### 项目深挖
14. 首屏加载怎么优化？从X秒到X秒怎么做的？
15. 组件库怎么设计的？怎么保证复用性？
16. 遇到过什么兼容性bug？怎么解决的？
---
### 算法工程师
#### 机器学习基础
1. 过拟合和欠拟合？怎么解决？
2. 特征工程怎么做？特征选择方法？
3. LR、SVM、GBDT原理？优缺点？
4. 模型评估指标？精确率召回率F1？ROC AUC？
#### 深度学习
5. CNN、RNN、LSTM原理？
6. Transformer原理？注意力机制？
7. BERT和GPT区别？预训练任务？
8. 模型压缩方法？量化剪枝蒸馏？
#### 推荐/NLP/CV（根据方向）
9. 推荐系统架构？召回排序重排？
10. 文本分类怎么做？
11. 目标检测/YOLO原理？
#### 工程实践
12. 模型训练怎么调参？
13. 模型怎么部署上线？
14. A/B测试怎么设计？
#### 项目深挖
15. 数据量多大？特征维度多少？
16. 模型效果怎么评估？和baseline对比？
17. 遇到过什么难题？怎么解决的？
---
## 五、设计类岗位
### UI设计
1. 请介绍一个你最有代表性的作品
2. 你的设计流程是什么？
3. 怎么做竞品分析？
4. 设计规范怎么制定？
5. 怎么保证设计还原度？
6. 设计和开发有分歧怎么办？
7. 怎么验证设计效果？
8. 最近关注的设计趋势？
9. 设计系统和组件库的区别？
10. 可访问性（a11y）你了解多少？
### UX/交互设计
1. 请介绍一个完整的设计案例（从研究到验证）
2. 用户研究有哪些方法？各适合什么场景？
3. 怎么做可用性测试？
4. 怎么写用户画像？
5. 信息架构怎么设计？
6. 怎么衡量UX设计的效果？
7. 设计决策被质疑怎么办？
8. 怎么平衡商业目标和用户体验？
---
## 六、数据类岗位
### 数据分析
1. SQL：查询每个用户的累计消费金额？
2. SQL：计算7日留存率？
3. SQL：窗口函数使用场景？
4. 怎么搭建指标体系？
5. 漏斗分析怎么做？
6. A/B测试样本量怎么计算？
7. 如果实验结果不显著怎么办？
8. DAU下降了怎么分析？
### 数据挖掘
1. 特征工程怎么做？
2. 怎么处理数据不平衡？
3. 模型怎么选择？
4. 怎么评估模型效果？
5. 模型怎么部署上线？
---
## 七、按项目类型的问题
### 增长类项目
- 增长的核心驱动力是什么？
- 怎么判断一个增长策略是否可持续？
- 怎么做增长实验？
### AI应用类项目
- 怎么评估AI生成内容的质量？
- AI幻觉问题怎么解决？
- 怎么设计人机结合的工作流？
### 交易类项目
- 交易漏斗怎么优化？
- 怎么提升支付转化率？
- 怎么处理退款/售后？
### 技术优化类项目
- 性能瓶颈怎么定位？
- 怎么设计高并发架构？
- 代码重构的原则？
---
## 八、反问环节建议问题
### 业务相关
1. 团队当前的核心目标和挑战是什么？
2. 这个岗位具体会负责哪些工作？
3. 团队的工作节奏和协作方式是怎样的？
### 成长相关
4. 对这个岗位的期望是什么？
5. 团队有什么培养机制？
6. 您认为这个岗位最重要的能力是什么？
### 流程相关
7. 接下来的面试流程是怎样的？
8. 大概什么时候会有反馈？
---
## 使用建议
1. **技术类岗位**：重点准备算法题（LeetCode）、系统设计、项目深挖
2. **设计类岗位**：重点准备作品集讲解、设计思路、设计验证
3. **产品/运营类岗位**：重点准备项目深挖、数据分析、开放题
4. **所有岗位**：准备3-5个核心项目，每个项目准备5-10个可能的问题
FILE:assets/templates/Java后端开发.md
# Java后端开发简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间] | [GitHub/技术博客链接]
**求职意向**：Java后端开发实习生 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段后端开发经验，熟悉高并发、分布式系统设计
- 熟练掌握Java技术栈（Spring Boot、MySQL、Redis、Kafka等），有[X万QPS]高并发系统开发经验
- 具备良好的代码规范意识，熟悉单元测试、Code Review等工程实践
---
## 技术栈
**编程语言**：Java（精通）、Python（熟悉）/ Go（了解）
**框架与中间件**：Spring Boot、Spring Cloud、MyBatis、Netty
**数据库**：MySQL（索引优化、分库分表）、Redis（缓存设计、分布式锁）、Elasticsearch
**消息队列**：Kafka、RabbitMQ、RocketMQ
**微服务与云原生**：Dubbo、gRPC、Docker、Kubernetes（了解）
**工具与工程实践**：Git、Maven、Jenkins、单元测试（JUnit）、Code Review、CI/CD
---
## 工作经历/项目经历
### [公司名] — [部门] Java后端开发实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，性能数据前置]**
- **项目背景**：[业务场景，系统规模，技术挑战]
- **我的职责**：负责[XX模块]的设计与开发，独立完成后端[X个接口]
- **技术方案**：
- [技术点1：架构设计/数据库设计/缓存策略]
- [技术点2：高并发处理/性能优化/分布式事务]
- [技术点3：代码优化/重构/工程规范]
- **核心成果**：
- 接口QPS从[X]提升至[X]，响应时间从[Xms]降至[Xms]
- 系统可用性达到[X个9]，支撑[X万DAU]
- 代码量减少[X%]，可维护性显著提升
**[项目名2] - [一句话成果]**
- ...
### [公司名2/个人项目] — Java后端开发
*[开始时间] - [结束时间]*
...
---
## 项目/开源贡献（可选）
### [项目名称] [GitHub链接]
*[角色] | [时间]*
- **项目描述**：[一句话介绍项目背景和目标]
- **技术栈**：[使用的技术]
- **我的贡献**：
- [具体贡献1：功能开发/性能优化/Bug修复]
- [具体贡献2：架构设计/代码重构/文档完善]
- **项目成果**：[Star数/Fork数/实际应用效果]
---
## 教育背景
**[学校名]** — [计算机/软件工程/相关专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：数据结构、算法、操作系统、计算机网络、数据库]
- [荣誉奖项：ACM竞赛、数学建模、优秀项目]
---
## 自我评价（可选）
[一句话概括技术能力]：熟悉Java技术栈，有高并发系统开发经验，注重代码质量和工程规范
[一句话表达特质]：对技术有热情，善于学习新技术，有良好的团队协作能力
---
## Java后端关键词
**基础**：Java基础、集合框架、多线程、JVM、GC、IO/NIO
**框架**：Spring、Spring Boot、Spring Cloud、MyBatis、Netty
**数据库**：MySQL（索引、事务、锁、慢查询优化）、Redis（缓存、分布式锁、持久化）
**中间件**：Kafka、RabbitMQ、RocketMQ、Elasticsearch、Zookeeper
**微服务**：服务注册发现、负载均衡、熔断降级、分布式事务、API网关
**工程实践**：单元测试、Code Review、Git工作流、CI/CD、敏捷开发
**系统设计**：高并发、高可用、分布式、CAP、BASE、最终一致性
---
## 技术类简历要点
### ✅ DO
- 技术栈明确写出具体框架和版本
- 性能数据量化（QPS、TPS、响应时间、可用性）
- 突出解决的技术难点
- 代码质量和工程实践
- GitHub/技术博客链接（如果有）
### ❌ DON'T
- 只列技术名词，没有实际项目应用
- 写"精通"（除非真的很精通）
- 写与岗位无关的技术（投Java岗写大量前端技术）
- 不写代码量/接口数/模块复杂度
---
## 示例对比
### ❌ 原写法
负责后端开发，使用Spring Boot开发了几个接口，优化了数据库查询。
### ✅ 优化写法
接口QPS提升200% | 负责XX系统核心模块开发，通过缓存优化和SQL优化提升系统性能
- 系统架构：设计并实现XX模块，提供X个RESTful API，支撑X万DAU
- 性能优化：
- 引入Redis缓存热点数据，数据库查询减少70%，接口响应从200ms降至80ms
- 优化慢查询SQL，添加联合索引，查询耗时从500ms降至50ms
- 使用Kafka异步处理削峰，系统QPS从1000提升至3000
- 代码质量：编写单元测试（覆盖率80%+），参与Code Review，保证代码质量
FILE:assets/templates/UI设计.md
# UI设计简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间] | [作品集链接/Dribbble/站酷]
**求职意向**：UI设计实习生 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段UI设计经验，熟悉移动端和Web端设计规范
- 具备完整的设计流程经验（需求分析→设计执行→设计验证），输出[X+]高质量界面
- 有设计系统/组件库搭建经验，注重设计一致性和复用性
---
## 设计工具
**界面设计**：Figma（主力）、Sketch、Adobe XD
**视觉设计**：Photoshop、Illustrator
**原型与交付**：Figma原型、Principle（动效）、蓝湖/即时设计（交付）
**设计管理**：Notion、MasterGo
---
## 工作经历/项目经历
### [公司名] — [部门] UI设计实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，设计成果前置]**
- **项目背景**：[业务场景、目标用户、设计目标]
- **我的职责**：负责[XX产品/功能]的UI设计，输出[X个页面/X个组件]
- **设计过程**：
- [需求分析：用户调研/竞品分析/设计目标]
- [设计执行：信息架构/交互流程/视觉设计]
- [设计验证：可用性测试/数据验证/迭代优化]
- **核心成果**：
- 设计[X个界面]，沉淀[X个组件]，覆盖[X个核心场景]
- 设计系统文档化，提升团队效率[X%]
- 用户满意度从[X]提升至[X]，转化率提升[X%]
**[项目名2] - [一句话成果]**
- ...
### [公司名2/个人项目] — UI设计
*[开始时间] - [结束时间]*
...
---
## 设计系统/组件库（可选）
### [设计系统名称]
*[角色] | [时间]*
- **设计范围**：[组件数量/页面模板数量/覆盖场景]
- **设计规范**：[色彩/字体/图标/间距/阴影等规范]
- **成果**：
- 组件复用率[X%]，设计效率提升[X%]
- 开发效率提升[X%]，减少沟通成本
---
## 教育背景
**[学校名]** — [设计/美术/视觉传达/相关专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：平面设计、交互设计、用户研究、设计心理学]
- [荣誉奖项：设计竞赛、奖学金、优秀毕业设计]
---
## 作品集亮点（可选）
- [作品1：项目类型，设计亮点，成果数据]
- [作品2：项目类型，设计亮点，成果数据]
- [更多作品见：作品集链接]
---
## 自我评价（可选）
[一句话概括设计能力]：具备完整的UI设计能力，熟悉设计系统和组件化思维，注重设计与业务的结合
[一句话表达特质]：对设计趋势保持关注，善于与产品、开发协作，追求用户体验与视觉美感的平衡
---
## UI设计关键词
**设计基础**：色彩理论、排版、字体、图标设计、视觉层次
**设计规范**：iOS HIG、Material Design、Ant Design、设计系统
**设计流程**：用户研究、竞品分析、信息架构、原型设计、可用性测试
**设计交付**：设计标注、切图、设计规范文档、设计走查
**设计思维**：用户中心、一致性、可用性、可访问性（a11y）
---
## UI简历要点
### ✅ DO
- 作品集链接（最重要！）
- 设计成果量化（页面数、组件数、效率提升）
- 设计验证数据（用户满意度、转化率）
- 设计系统/组件库经验
- 完整的设计流程展示
### ❌ DON'T
- 不写作品集链接
- 只写"负责界面设计"没有设计细节
- 不写设计思考和设计目标
- 设计文件管理混乱
---
## 示例对比
### ❌ 原写法
负责App的UI设计，设计了首页、个人中心等页面。
### ✅ 优化写法
设计系统覆盖30+页面 | 负责XX App UI设计，搭建组件库提升团队效率
- 需求分析：参与用户调研和竞品分析，确定设计目标和风格定位
- 界面设计：
- 完成首页、搜索、详情、个人中心等12个核心页面设计
- 设计图标50+，统一视觉语言
- 适配iOS/Android双端，确保设计一致性
- 组件沉淀：
- 搭建UI组件库，沉淀30+可复用组件（按钮、卡片、列表等）
- 输出设计规范文档，设计效率提升40%，开发还原度95%+
- 设计验证：
- 可用性测试发现5个体验问题并优化
- 上线后用户满意度从3.8提升至4.5（5分制）
FILE:assets/templates/UX设计.md
# UX/交互设计简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间] | [作品集链接]
**求职意向**：UX设计实习生 / 交互设计实习生 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段UX/交互设计经验，熟悉用户研究方法和交互设计流程
- 具备从用户研究到设计验证的完整项目经验，有[X个]完整案例
- 注重数据验证，通过用户测试和数据分析优化设计方案
---
## 设计工具与方法
**设计工具**：Figma、Sketch、Axure、Principle（动效）
**用户研究**：用户访谈、问卷调查、可用性测试、A/B测试
**设计方法**：Personas、User Journey Map、信息架构、任务流程、原型设计
**设计验证**：数据分析、可用性测试、A/B测试、NPS调研
---
## 工作经历/项目经历
### [公司名] — [部门] UX设计实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，体验指标前置]**
- **项目背景**：[业务场景、用户痛点、设计目标]
- **用户研究**：
- [方法：用户访谈X人/问卷X份/数据分析]
- [发现：核心痛点X个/用户行为特征]
- **设计过程**：
- [信息架构： restructuring/优化导航]
- [交互设计：核心流程优化/新功能设计]
- [原型验证：低保真/高保真原型，X轮迭代]
- **设计验证**：
- [可用性测试：X名用户，发现X个问题，优化X处]
- [数据验证：核心指标从X提升至X]
- **核心成果**：
- 任务完成率从[X%]提升至[X%]
- 用户满意度从[X]提升至[X]
- 功能使用率提升[X%]
**[项目名2] - [一句话成果]**
- ...
### [公司名2/个人项目] — UX设计
*[开始时间] - [结束时间]*
...
---
## 设计案例（可选）
### [案例名称] [链接]
*[角色] | [时间]*
- **设计问题**：[一句话描述]
- **研究方法**：[用户访谈/数据分析/竞品分析]
- **设计方案**：[核心设计决策]
- **验证结果**：[数据证明]
---
## 教育背景
**[学校名]** — [交互设计/人机交互/心理学/相关专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：人机交互、用户研究、设计心理学、信息架构]
- [荣誉奖项：设计竞赛、优秀项目]
---
## 自我评价（可选）
[一句话概括设计能力]：具备完整的UX设计能力，擅长用户研究和数据验证，注重设计方案的业务价值
[一句话表达特质]：对用户有同理心，善于发现和定义问题，用设计解决真实用户需求
---
## UX关键词
**用户研究**：用户访谈、问卷调查、焦点小组、用户画像、场景分析
**设计方法**：用户旅程图、服务蓝图、信息架构、任务流程、线框图、原型
**设计验证**：可用性测试、A/B测试、眼动测试、数据分析、NPS
**设计思维**：以用户为中心、设计冲刺、敏捷设计、迭代优化
---
## UX简历要点
### ✅ DO
- 突出用户研究过程和方法
- 设计决策基于用户洞察
- 设计验证数据（可用性测试、数据指标）
- 完整的设计流程展示
- 作品集包含研究过程和设计方案
### ❌ DON'T
- 只展示最终设计稿，不写设计过程
- 不写用户研究和验证方法
- 设计决策没有依据
- 不量设计效果
---
## 示例对比
### ❌ 原写法
负责App的交互设计，优化了下单流程。
### ✅ 优化写法
任务完成率提升25% | 通过用户研究和流程优化，显著提升下单体验
- 问题发现：
- 用户访谈10人，发现下单流程存在3大痛点（步骤多、信息杂、反馈弱）
- 数据分析：漏斗分析显示，67%用户在第3步流失
- 设计优化：
- 信息重组：合并3个步骤为2个，减少用户认知负担
- 进度可视：增加步骤指示器，提升用户掌控感
- 反馈强化：关键操作增加即时反馈，降低焦虑
- 验证迭代：
- 可用性测试5人，任务完成率从60%提升至85%
- A/B测试上线，整体转化率提升15%，GMV增长X万/月
FILE:assets/templates/产品经理.md
# 产品经理简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间]
**求职意向**：产品实习生 - [C端/B端/增长/策略]方向 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段产品实习经验，独立负责[XX]模块从0到1设计与落地
- 擅长[用户研究/数据分析/增长策略]，有完整的[需求分析→方案设计→数据验证]闭环经验
- [具体亮点，如：推动XX项目上线，DAU提升XX% / 主导AI应用落地，替代X人效]
---
## 工作经历
### [公司名] — [部门] 产品实习生（[方向]）
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，数据前置]**
- **项目背景**：[为什么要做，解决什么问题，目标用户是谁]
- **我的职责**：独立负责[XX]模块，输出[PRD/原型/数据方案]，协调[XX/XX/XX]团队推进
- **核心动作**：
- [动作1：需求分析/用户调研，具体方法]
- [动作2：方案设计，关键决策点]
- [动作3：项目推动，如何协调资源]
- **项目成果**：[量化数据1]，[量化数据2]，[定性成果]
**[项目名2] - [一句话成果]**
- ...
### [公司名2] — [部门] 产品实习生
*[开始时间] - [结束时间]*
...
---
## 项目/研究经历（可选）
### [项目名]
*[角色] | [时间]*
- [STAR法则描述，突出产品思维和数据能力]
- [使用的工具和方法]
- [成果和影响]
---
## 教育背景
**[学校名]** — [专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：产品相关、数据相关、技术相关]
- [荣誉奖项：奖学金、竞赛、论文]
---
## 技能特长
**产品工具**：Axure、Figma、墨刀、MasterGo（独立完成[X]个PRD）
**数据分析**：SQL（多表联查、窗口函数）、Excel（数据透视、复杂函数）、[Python/R基础]
**方法论**：用户调研（访谈/问卷）、A/B测试、漏斗分析、用户分层、[AARRR/RFM等模型]
**其他**：[英语能力]、[技术理解]、[行业认知]
---
## 自我评价（可选，1-2句话）
[一句话概括核心优势]：[具体证据]
[一句话表达求职动机]：对[XX领域]有强烈兴趣，希望在[目标公司]深耕[XX方向]
---
## 写作要点提示
### ✅ DO
- 数据前置：每个项目第一句放核心数据
- 动词开头：负责、设计、推动、协调、搭建
- 量化成果：用户数、转化率、收入、效率提升
- 体现ownership：独立负责、主导、owner
### ❌ DON'T
- 避免"参与""协助"等弱化词，除非确实是支持角色
- 避免纯描述性语言，要有数据和成果
- 避免一页纸以上（校招/实习）
- 避免错别字和格式不统一
---
## 示例对比
### ❌ 原写法
负责用户增长策略，通过活动策划提升了用户量，与其他部门协作保证项目上线。
### ✅ 优化写法
用户增长XX万（+XX%）| 主导增长策略设计与落地，策划XX活动，通过裂变+投放组合实现用户规模突破
- 策略设计：拆解增长漏斗，识别关键转化节点，设计针对性的裂变激励机制
- 资源整合：协调市场、技术、设计3个团队，2周内完成从方案到上线
- 数据验证：搭建实时监控看板，根据数据及时调整投放策略，ROI提升XX%
FILE:assets/templates/产品运营.md
# 产品运营简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间]
**求职意向**：产品运营实习生 - [用户运营/内容运营/活动运营/增长]方向 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段运营实习经验，负责[DAU XX万/GMV XX万]量级的[产品/活动/内容]运营
- 擅长[用户增长/活动策划/内容策略]，有从0到1搭建[运营体系/活动玩法/内容生态]的经验
- 数据驱动决策，通过[数据分析/用户调研]优化策略，实现[XX指标提升XX%]
---
## 工作经历
### [公司名] — [部门] 运营实习生（[方向]）
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，数据前置]**
- **运营目标**：[DAU/留存/GMV/内容量]目标，目标用户[XX群体]
- **策略设计**：
- [策略1：增长策略，如何拉新]
- [策略2：留存策略，如何促活]
- [策略3：变现策略，如何转化]
- **执行落地**：
- [动作1：具体做了什么事]
- [动作2：协调了哪些资源]
- [动作3：用了什么工具/方法]
- **数据成果**：[核心指标1]提升XX%，[核心指标2]达到XX，超预期XX%
**[项目名2] - [一句话成果]**
- ...
### [公司名2] — [部门] 运营实习生
*[开始时间] - [结束时间]*
...
---
## 运营项目/案例（可选）
### [个人账号/社团/比赛项目]
*[角色] | [时间]*
- [即使是非实习经历，也可体现运营能力]
- [如：个人小红书账号，X个月涨粉X万，变现X元]
- [如：社团活动，策划XX活动，参与人数X人]
---
## 教育背景
**[学校名]** — [专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：市场营销、消费者行为、统计学]
- [荣誉奖项：奖学金、竞赛、运营相关证书]
---
## 技能特长
**数据分析**：SQL（数据提取、多维度分析）、Excel（数据透视、VLOOKUP）、[Tableau/FineBI]
**运营工具**：神策/GrowingIO、[内容创作工具]、[社群运营工具]
**内容能力**：文案撰写、活动策划、用户分层、[短视频/图文创作]
**方法论**：AARRR增长模型、RFM用户分层、漏斗分析、A/B测试
---
## 自我评价（可选）
[一句话概括运营能力]：擅长[XX]运营，有[XX]量级的实操经验
[一句话表达特质]：对数据敏感，有创意，执行力强，善于从用户视角发现问题
---
## 运营岗位关键词
**用户运营**：拉新、留存、促活、LTV、用户分层、用户画像、社群运营、用户反馈
**内容运营**：内容策划、创作者运营、内容分发、爆款打造、UGC/PGC、内容审核
**活动运营**：活动策划、裂变增长、GMV、转化率、ROI、爆款活动、节日营销
**增长运营**：增长黑客、渠道投放、裂变、病毒传播、冷启动、DAU、留存率
---
## 示例对比
### ❌ 原写法
负责活动运营工作，策划了一些线上活动，提升了用户活跃度。
### ✅ 优化写法
DAU提升XX% | 策划并落地XX活动，通过[裂变机制/奖励设计]实现用户规模突破
- 活动策划：设计[XX玩法]，设置[XX奖励机制]，降低参与门槛提升传播率
- 渠道投放：在[XX渠道]投放，CTR[X%]，获客成本[X元]，ROI[X]
- 数据复盘：活动期间监控[XX指标]，根据实时数据调整[XX策略]，最终[XX成果]
FILE:assets/templates/前端开发.md
# 前端开发简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间] | [GitHub/个人网站]
**求职意向**：前端开发实习生 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段前端开发经验，熟悉现代前端技术栈和工程化实践
- 熟练掌握[React/Vue]，有大型SPA应用开发经验，注重性能优化和用户体验
- 具备组件化思维，有组件库/工具库开发经验
---
## 技术栈
**核心框架**：React（Hooks、Redux/Mobx）/ Vue（3.0、Pinia/Vuex）
**语言与基础**：JavaScript（ES6+）、TypeScript、HTML5、CSS3
**工程化工具**：Webpack、Vite、Rollup、Babel、ESLint/Prettier
**UI框架**：Ant Design、Element Plus、Tailwind CSS
**可视化**：ECharts、D3.js、Canvas（了解）
**移动端**：React Native、微信小程序、Taro（了解）
**测试与质量**：Jest、React Testing Library、单元测试、E2E测试
**工具**：Git、npm/yarn/pnpm、Chrome DevTools
---
## 工作经历/项目经历
### [公司名] — [部门] 前端开发实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，性能/体验数据前置]**
- **项目背景**：[业务场景，项目规模，技术栈]
- **我的职责**：负责[XX模块/页面]的开发，实现[X个页面/X个组件]
- **技术亮点**：
- [技术点1：组件设计/状态管理/路由优化]
- [技术点2：性能优化（首屏加载/渲染优化/代码分割）]
- [技术点3：工程化（构建优化/代码规范/自动化）]
- **核心成果**：
- 首屏加载时间从[Xs]降至[Xs]，性能评分从[X]提升至[X]
- 代码复用率提升[X%]，开发效率提升[X%]
- 用户满意度提升[X%]，页面转化率提升[X%]
**[项目名2] - [一句话成果]**
- ...
### [公司名2/个人项目] — 前端开发
*[开始时间] - [结束时间]*
...
---
## 开源项目/技术博客（可选）
### [项目名称] [GitHub链接]
*[角色] | [技术栈]*
- [项目简介：一句话说明]
- [技术亮点：组件设计/性能优化/工程化实践]
- [Star数/下载量/实际应用]
---
## 教育背景
**[学校名]** — [计算机/软件工程/设计相关专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：数据结构、算法、人机交互、Web开发]
- [荣誉奖项：前端竞赛、黑客松、优秀项目]
---
## 自我评价（可选）
[一句话概括技术能力]：熟悉现代前端技术栈，注重用户体验和性能优化，有组件化开发经验
[一句话表达特质]：对前端新技术保持学习热情，关注用户体验，注重代码质量
---
## 前端关键词
**框架**：React、Vue、Angular、Svelte
**基础**：ES6+、TypeScript、HTML5、CSS3、Flex/Grid、响应式设计
**工程化**：Webpack、Vite、Rollup、Babel、npm/yarn/pnpm、Monorepo
**性能**：首屏优化、懒加载、代码分割、Tree Shaking、CDN、缓存策略
**体验**：动画、交互设计、无障碍（a11y）、PWA
**测试**：Jest、Testing Library、Cypress、单元测试、集成测试
**可视化**：ECharts、D3.js、Canvas、WebGL（了解）
---
## 前端简历要点
### ✅ DO
- 明确写出主要技术栈（React还是Vue）
- 性能数据量化（首屏时间、包体积、渲染性能）
- 突出用户体验优化
- 组件化/工程化实践
- GitHub/个人网站链接（展示作品）
### ❌ DON'T
- 罗列所有技术但不写实际应用
- 只写"负责页面开发"没有技术细节
- 不写性能数据和用户体验指标
---
## 示例对比
### ❌ 原写法
使用Vue开发了后台管理系统的前端页面，实现了一些功能。
### ✅ 优化写法
首屏加载优化60% | 负责XX后台管理系统前端开发，通过工程化实践提升开发效率和用户体验
- 技术架构：基于Vue3 + TypeScript搭建项目，实现X个页面和X个复用组件
- 性能优化：
- 路由懒加载+组件异步加载，首屏时间从3s降至1.2s
- Webpack优化（代码分割、Tree Shaking），打包体积减少40%
- 虚拟滚动优化大数据列表，渲染10万条数据无卡顿
- 工程化：
- 搭建组件库，沉淀20+业务组件，代码复用率提升50%
- 配置ESLint+Prettier规范，接入Git Hooks保证代码质量
- 编写单元测试（覆盖率70%+），提升代码稳定性
FILE:assets/templates/商业化产品.md
# 商业化产品简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间]
**求职意向**：商业化产品实习生 - [广告/变现/会员/交易]方向 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段商业化产品/运营经验，负责[DAU XX万/GMV XX亿]产品的变现策略
- 熟悉[广告竞价/会员体系/交易抽成]等商业模式，有完整的变现策略设计→上线→优化经验
- 通过[策略优化/实验设计]提升[广告收入/ARPU/转化率]，实现[收入提升XX%]
---
## 工作经历
### [公司名] — [部门] 商业化产品实习生（[方向]）
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，收入数据前置]**
- **商业目标**：[收入目标]，商业模式[广告/会员/交易/其他]，目标用户[XX群体]
- **策略设计**：
- [变现策略1：广告位设计/竞价策略/定向优化]
- [变现策略2：会员权益设计/定价策略]
- [变现策略3：交易抽成/佣金策略]
- **实验验证**：
- 设计A/B测试验证策略效果
- 实验组[XX] vs 对照组[XX]，[收入/ARPU/转化率]提升XX%，统计显著
- **核心成果**：[收入提升XX%]，[ARPU提升XX%]，[广告主/用户满意度]
**[项目名2] - [一句话成果]**
- ...
### [公司名2] — [部门] 商业化产品实习生
*[开始时间] - [结束时间]*
...
---
## 项目/创业经历（可选）
### [创业项目/商业分析项目]
*[角色] | [时间]*
- [即使没有商业化实习，也可体现商业敏感度]
- [如：分析XX商业模式，提出XX变现建议]
- [如：个人副业/小生意，营收X元]
---
## 教育背景
**[学校名]** — [专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：经济学、市场营销、博弈论、会计学]
- [荣誉奖项：商业计划书比赛、案例分析大赛]
---
## 技能特长
**商业分析**：商业模式分析、竞品分析、定价策略、ROI分析、LTV/CAC计算
**数据分析**：SQL、Excel、A/B测试设计、漏斗分析、收入归因
**广告产品**：竞价广告、品牌广告、信息流广告、广告归因、DMP/DSP基础
**变现模式**：广告变现、会员订阅、交易抽成、增值服务
---
## 自我评价（可选）
[一句话概括商业能力]：兼具商业敏感度和产品思维，擅长在用户体验和商业收入之间找平衡
[一句话表达特质]：对数字敏感，有成本意识，善于从商业视角思考产品决策
---
## 商业化产品关键词
**广告变现**：CPM、CPC、CPA、RTB、竞价排序、eCPM、广告填充率、广告质量度
**会员变现**：订阅模式、权益设计、定价策略、续费率、LTV、付费转化率
**交易变现**：GMV、抽成比例、佣金、交易手续费、商家分层
**核心指标**：ARPU、ARPPU、广告收入、变现效率、商业化率、ROI
---
## 示例对比
### ❌ 原写法
负责广告产品优化，提升了广告收入，优化了广告体验。
### ✅ 优化写法
广告收入提升XX% | 优化信息流广告竞价策略，平衡收入与用户体验
- 问题诊断：通过数据分析发现[XX广告位]eCPM低（X元），填充率仅X%，存在[XX问题]
- 策略设计：优化竞价排序算法，引入[XX因素]；调整广告位样式，提升CTR[X%]
- 实验验证：A/B测试显示，实验组收入提升[X%]，用户停留时长无显著下降，全量上线
- 长期优化：建立广告质量度评估体系，长期优化广告主结构
FILE:assets/templates/数据分析.md
# 数据分析简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间]
**求职意向**：数据分析实习生 | [校招/暑期/日常]实习 | [SQL/Python/R]
---
## 核心优势
- [X]段数据分析实习经验，支撑[DAU XX万/GMV XX亿]量级业务的数据决策
- 熟练使用SQL进行复杂数据提取和清洗，掌握[Python/R]进行统计分析和建模
- 搭建[XX]数据监控体系，通过数据分析发现[XX问题]，推动[XX优化]，实现[XX提升]
---
## 工作经历
### [公司名] — [部门] 数据分析实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，数据前置]**
- **分析目标**：[业务问题]，目标[优化XX指标/发现XX规律/验证XX假设]
- **数据处理**：
- [数据来源：数据库/埋点/第三方]
- [数据清洗：处理缺失值、异常值、重复数据]
- [数据关联：XX表与XX表关联，字段映射]
- **分析方法**：
- [分析方法1：描述统计/漏斗分析/留存分析/归因分析]
- [分析方法2：A/B测试设计/用户分层/聚类分析]
- [分析方法3：预测模型/回归分析/时间序列]
- **核心发现**：[发现XX规律/问题]，[XX因素]对[XX指标]影响最大
- **业务建议**：建议[XX优化措施]，预期[XX效果]，实际落地后[XX成果]
**[项目名2] - [一句话成果]**
- ...
### [公司名2] — [部门] 数据分析实习生
*[开始时间] - [结束时间]*
...
---
## 项目/研究经历
### [项目名称]
*[角色] | [时间]*
- **项目背景**：[业务场景/研究问题]
- **数据处理**：[使用的数据集、数据清洗过程]
- **分析方法**：[统计方法/机器学习模型]
- **核心结论**：[数据分析发现的关键结论]
- **可视化**：[使用的工具，如图表类型]
---
## 教育背景
**[学校名]** — [统计学/数学/计算机/数据科学/计量经济]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [核心课程：统计学、概率论、回归分析、机器学习、数据库]
- [荣誉奖项：数学建模竞赛、数据分析比赛、优秀论文]
---
## 技能特长
**SQL**：复杂查询（多表JOIN、子查询、窗口函数）、性能优化、Hive/Spark SQL
**Python/R**：
- Python：pandas（数据处理）、numpy（数值计算）、matplotlib/seaborn（可视化）、sklearn（机器学习基础）
- R：[如果会的话，描述能力]
**统计分析**：假设检验、置信区间、回归分析、A/B测试设计、显著性检验
**可视化工具**：Tableau、FineBI、PowerBI、[Excel高级图表]
**其他**：[机器学习基础]、[数据仓库知识]、[埋点设计经验]
---
## 自我评价（可选）
[一句话概括分析能力]：擅长将业务问题转化为数据问题，用数据驱动业务决策
[一句话表达特质]：逻辑严谨，对数据敏感，善于从复杂数据中发现业务洞察
---
## 数据分析岗位关键词
**数据提取**：SQL、Hive、Spark、数据仓库、ETL、数据清洗
**分析方法**：漏斗分析、留存分析、路径分析、归因分析、用户分层、RFM、 cohort分析
**实验设计**：A/B测试、假设检验、显著性、样本量计算、实验分组、效果评估
**指标体系**：指标拆解、北极星指标、虚荣指标、埋点设计、数据看板
**统计建模**：回归分析、聚类分析、预测模型、特征工程、模型评估
---
## 示例对比
### ❌ 原写法
负责数据分析工作，写SQL查询数据，制作报表提供给业务方。
### ✅ 优化写法
通过数据分析发现[XX问题]，推动[XX优化]，实现[XX指标提升XX%]
- 问题发现：通过漏斗分析发现[XX环节]转化率异常（仅X%，低于大盘X%），定位[XX原因]
- 深度分析：下钻到[XX维度]，发现[XX用户群体/场景]问题最集中，占比XX%
- 策略建议：建议优化[XX]，预期提升[X%]，经A/B测试验证，实际提升[X%]，全量上线
FILE:assets/templates/新媒体运营.md
# 新媒体运营简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间] | [公众号/抖音/小红书账号]
**求职意向**：新媒体运营实习生 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段新媒体运营经验，熟悉[抖音/小红书/公众号/视频号]等平台运营
- 擅长内容策划和用户增长，[个人账号/操盘账号]粉丝[X万]，爆款内容[X条]
- 具备内容创作能力，能独立完成[文案/拍摄/剪辑/设计]
---
## 工作经历/项目经历
### [公司名] — [部门] 新媒体运营实习生
*[开始时间] - [结束时间]*
**[平台名称]运营 - [一句话成果，数据前置]**
- **账号定位**：[目标用户/内容方向/人设定位]
- **内容运营**：
- [内容规划：选题库搭建/内容日历/热点追踪]
- [内容生产：文案撰写/视频制作/图文设计]
- [产出数据：发文X篇/视频X条，平均播放量/阅读量X]
- **用户增长**：
- [增长策略：内容爆款/投放推广/裂变活动/跨平台引流]
- [增长数据：粉丝从X增长至X（+X%），爆款内容X条]
- **变现/转化**：
- [变现方式：广告/带货/引流/品牌合作]
- [变现数据：GMV X万/广告收入X万/线索X条]
- **核心成果**：
- 粉丝增长[X万]，[月/总]播放量/阅读量[X万]
- 爆款内容[X条]（播放量/阅读量超[X万]）
- [其他成果：账号获奖/品牌合作/引流转化]
**[项目名2] - [一句话成果]**
- ...
### [个人账号/副业] — 自媒体运营
*[开始时间] - [结束时间]*
- **平台**：[抖音/小红书/B站/公众号]
- **定位**：[内容方向]
- **成果**：粉丝[X万]，[爆款内容X条]，[变现X元]
---
## 教育背景
**[学校名]** — [新闻传播/中文/市场营销/相关专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：传播学、新闻写作、市场营销、网络与新媒体]
- [荣誉奖项：新媒体大赛、写作比赛]
---
## 技能特长
**内容创作**：文案撰写、短视频脚本、拍摄剪辑、图文排版、基础设计
**平台运营**：抖音/小红书/公众号/视频号/微博/B站等平台规则和算法
**数据分析**：新榜/蝉妈妈/灰豚等数据工具，内容数据分析
**工具**：剪映/PR/PS/Canva、秀米/135编辑器、创客贴
---
## 新媒体运营关键词
**内容方向**：短视频、图文、直播、Vlog、知识付费、种草、测评
**增长方法**：爆款打造、热点借势、算法优化、投放推广、裂变引流
**变现模式**：广告植入、直播带货、知识付费、私域引流、品牌合作
**核心指标**：粉丝数、播放量、互动率、完播率、转化率、GMV
**平台特性**：抖音算法、小红书种草、公众号私域、视频号社交
---
## 示例对比
### ❌ 原写法
负责公司抖音账号运营，发布了一些视频，粉丝增长还不错。
### ✅ 优化写法
抖音粉丝增长10万（+200%）| 从0到1搭建公司抖音账号，打造3条百万播放爆款
- 账号定位：分析目标用户（XX岁+XX画像），确定"XX"内容方向和人设
- 内容生产：
- 搭建选题库（100+选题），保证日更不断更
- 独立完成脚本、拍摄、剪辑，月均产出30条视频
- 追踪热点，快速响应，3条视频播放量超100万
- 增长策略：
- 优化标题、封面、发布时间，平均播放量从5000提升至5万
- DOU+投放测试，ROI 1:3，低成本精准获客
- 直播引流，单场直播涨粉5000+
- 变现成果：引流私域[X人]，转化GMV[X万]，广告合作[X单]
FILE:assets/templates/用户运营.md
# 用户运营简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间]
**求职意向**：用户运营实习生 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段用户运营经验，熟悉用户生命周期管理和精细化运营
- 擅长用户分层和分群运营，通过[XX策略]提升[留存率/活跃度/转化率]
- 具备数据分析能力，能用数据驱动运营决策
---
## 工作经历/项目经历
### [公司名] — [部门] 用户运营实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，用户指标前置]**
- **运营目标**：[提升留存/促活/转化]，目标用户[XX群体]
- **用户分析**：
- [用户分层：RFM模型/生命周期/价值分层]
- [用户画像：特征分析/行为分析/需求洞察]
- **运营策略**：
- [策略1：针对X类用户的触达策略]
- [策略2：权益设计/活动设计/内容策略]
- [策略3：多渠道触达：Push/短信/站内信/社群]
- **核心成果**：
- [留存率/活跃度/转化率]从[X]提升至[X]，提升[X%]
- 用户LTV提升[X%]， churn率降低[X%]
- 运营成本降低[X%]，ROI提升[X]
**[项目名2] - [一句话成果]**
- ...
### [公司名2] — 用户运营实习生
*[开始时间] - [结束时间]*
...
---
## 教育背景
**[学校名]** — [专业]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：市场营销、消费者行为、统计学、数据分析]
- [荣誉奖项：奖学金、竞赛]
---
## 技能特长
**用户运营**：用户分层、生命周期管理、RFM模型、用户画像、触达策略
**数据分析**：SQL、Excel、漏斗分析、留存分析、LTV计算
**工具**：神策/GrowingIO、CRM系统、MA营销自动化工具
**运营方法**：AARRR模型、用户旅程地图、增长黑客
---
## 用户运营关键词
**用户分层**：RFM、生命周期、价值分层、用户画像、用户标签
**运营策略**：留存策略、促活策略、防流失策略、召回策略、转化策略
**触达渠道**：Push通知、短信、邮件、站内信、社群、客服
**核心指标**：留存率、活跃度、LTV、CAC、churn rate、NPS
**分析方法**：漏斗分析、路径分析、cohort分析、A/B测试
---
## 示例对比
### ❌ 原写法
负责用户运营工作，做了一些用户维护和活动策划。
### ✅ 优化写法
用户留存率提升15% | 通过用户分层和精细化运营，显著提升用户留存和活跃
- 用户分层：基于RFM模型将用户分为8类，识别高价值用户和流失风险用户
- 策略设计：
- 新用户：设计7日留存任务体系，新用户7日留存从30%提升至45%
- 沉默用户：设计召回活动，沉默用户唤醒率20%，回流成本降低30%
- 高价值用户：搭建VIP权益体系，高价值用户LTV提升25%
- 数据监控：搭建用户行为监控看板，实时追踪核心指标，及时调整策略
FILE:assets/templates/策略产品.md
# 策略产品简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间]
**求职意向**：策略产品实习生 - [推荐策略/搜索策略/定价策略/AI策略]方向 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段策略产品经验，负责[推荐/搜索/定价/AI]策略的设计与优化
- 熟悉[召回/排序/重排]策略链路，有[特征工程/模型评估/策略迭代]实战经验
- 通过[策略优化]提升[点击率/转化率/用户满意度]，实现[XX指标提升XX%]
---
## 工作经历
### [公司名] — [部门] 策略产品实习生（[方向]）
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，策略指标前置]**
- **策略目标**：[优化XX指标]，策略类型[推荐/搜索/定价/AI]，影响面[DAU XX万]
- **问题分析**：
- 现状：[当前策略存在的问题]
- 数据分析：[通过数据发现XX问题]，[XX场景/用户群体]问题最突出
- 原因定位：[问题根因是XX]
- **策略设计**：
- [策略1：召回层优化，XX]
- [策略2：排序层优化，引入XX特征/调整XX权重]
- [策略3：重排层优化，XX]
- **实验验证**：
- A/B测试设计：实验组[XX策略] vs 对照组[原策略]
- 核心指标：[点击率/转化率]提升XX%，统计显著
- 护栏指标：[多样性/新颖性/用户满意度]无显著下降
- **核心成果**：[指标1提升XX%]，[指标2提升XX%]，[其他收益]
**[项目名2] - [一句话成果]**
- ...
### [公司名2] — [部门] 策略产品实习生
*[开始时间] - [结束时间]*
...
---
## AI/算法项目（可选）
### [Prompt工程/模型应用项目]
*[角色] | [时间]*
- [AI策略相关经验]
- [如：设计Prompt实现XX功能，替代X人效]
- [如：参与大模型微调/应用，提升XX效果]
---
## 教育背景
**[学校名]** — [计算机/软件/数学/数据科学]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [核心课程：机器学习、数据结构、算法、统计学、最优化]
- [荣誉奖项：算法竞赛、数据挖掘比赛、论文]
---
## 技能特长
**策略设计**：召回策略、排序策略、重排策略、特征工程、策略评估
**算法基础**：机器学习基础、推荐算法原理（协同过滤/深度学习）、搜索算法基础
**AI应用**：Prompt工程、LLM应用、模型评估（准确率/召回率/F1）、AI工作流设计
**数据分析**：SQL、Python（pandas/sklearn基础）、A/B测试、实验设计
**工具**：[特征平台]、[模型服务平台]、[实验平台]
---
## 自我评价（可选）
[一句话概括策略能力]：兼具业务理解和算法思维，擅长将业务目标转化为可优化的策略问题
[一句话表达特质]：逻辑严谨，数据驱动，善于在复杂系统中找到关键优化点
---
## 策略产品关键词
**推荐策略**：召回（协同过滤/向量召回）、排序（LR/GBDT/深度学习）、重排（多样性/新颖性）
**搜索策略**：Query理解、意图识别、相关性排序、CTR预估
**特征工程**：用户画像、物品特征、上下文特征、交叉特征
**模型评估**：准确率、召回率、F1-score、AUC、ROC、混淆矩阵
**实验设计**：A/B测试、分组策略、样本量计算、显著性检验、AA测试
---
## 示例对比
### ❌ 原写法
负责推荐策略优化，调整了推荐算法，提升了点击率。
### ✅ 优化写法
点击率提升XX% | 优化推荐排序策略，引入XX特征，提升长尾内容分发效率
- 问题诊断：通过数据分析发现[XX类型内容]CTR低（X%），用户反馈"推荐重复"
- 策略设计：排序层引入[XX特征]（内容新鲜度、用户兴趣多样性），调整[XX权重]
- 实验验证：A/B测试，实验组CTR提升[X%]，多样性指标提升[X%]，用户满意度提升[X%]
- 长期优化：建立特征效果监控体系，持续迭代特征权重
FILE:assets/templates/算法工程师.md
# 算法工程师简历模板
## 个人信息
[姓名] | [电话] | [邮箱] | [学校-学历-毕业时间] | [GitHub/论文链接]
**求职意向**：算法工程师实习生 - [机器学习/NLP/CV/推荐]方向 | [校招/暑期/日常]实习
---
## 核心优势
- [X]段算法实习/研究经验，熟悉[机器学习/深度学习/NLP/CV/推荐]算法
- 在[XX会议/期刊]发表论文[X篇]，Kaggle/天池等竞赛[Top X%]成绩
- 熟练掌握Python、PyTorch/TensorFlow，有大规模数据处理和模型训练经验
---
## 技术栈
**编程语言**：Python（精通）、C++（熟悉）、SQL
**深度学习框架**：PyTorch、TensorFlow、Hugging Face Transformers
**机器学习库**：Scikit-learn、XGBoost、LightGBM、NumPy、Pandas
**NLP/CV工具**：NLTK、spaCy、OpenCV（根据方向选择）
**大数据工具**：Spark、Hive、Flink（了解）
**模型部署**：ONNX、TensorRT、Flask/FastAPI（模型 serving）
**开发工具**：Git、Linux、Jupyter、Wandb/TensorBoard
---
## 工作经历/研究经历
### [公司名] — [部门] 算法工程师实习生
*[开始时间] - [结束时间]*
**[项目名] - [一句话成果，模型效果前置]**
- **项目背景**：[业务场景、数据规模、模型挑战]
- **我的职责**：负责[XX模型]的设计与优化，提升[XX指标]
- **技术方案**：
- [数据处理：数据清洗、特征工程、样本构建]
- [模型设计：模型选型、网络结构、损失函数]
- [训练优化：训练策略、超参调优、模型融合]
- **核心成果**：
- [指标1]从[X]提升至[X]，相对提升[X%]
- [指标2]达到[X]，超过baseline [X%]
- 模型部署上线，支撑[X QPS/X DAU]
**[项目名2] - [一句话成果]**
- ...
### [学校实验室/科研项目]
*[角色] | [时间]*
**[研究方向]**
- 研究内容：[一句话概括研究问题和方法]
- 创新点：[方法创新/理论创新/应用创新]
- 论文成果：发表在[XX会议/期刊]，[CCF A/B/C类]
- 代码开源：[GitHub链接]
---
## 竞赛/开源项目（可选）
### [竞赛名称] [排名]
*[时间]*
- 赛题：[一句话描述]
- 方案：[核心技术点]
- 成绩：[Top X% / 第X名]
### [开源项目] [GitHub链接]
*[角色]*
- [项目简介]
- [Star数/Fork数]
---
## 教育背景
**[学校名]** — [计算机/人工智能/数学/统计]（[学位]）
*[入学时间] - [毕业时间] | GPA: X.XX/4.0*
- [相关课程：机器学习、深度学习、概率论、最优化、算法]
- [荣誉奖项：国家奖学金、优秀论文、竞赛获奖]
---
## 论文/专利（可选）
1. **[论文标题]**，[作者排名]，[会议/期刊名称]，[CCF分类]，[年份]
2. **[专利名称]**，[专利类型]，[状态：授权/申请中]，[年份]
---
## 自我评价（可选）
[一句话概括算法能力]：熟悉[XX方向]算法，有模型研发和落地经验，注重算法效果和工程实践结合
[一句话表达特质]：对算法研究有热情，善于从业务问题中抽象算法模型
---
## 算法工程师关键词
**机器学习**：LR、SVM、GBDT、XGBoost、LightGBM、特征工程、模型融合
**深度学习**：CNN、RNN、LSTM、Transformer、BERT、GPT、注意力机制
**NLP**：文本分类、序列标注、文本生成、预训练模型、Prompt工程
**CV**：图像分类、目标检测、图像分割、OCR（根据方向）
**推荐**：协同过滤、矩阵分解、DeepFM、Wide&Deep、序列推荐
**工程实践**：模型训练、超参调优、模型压缩、模型部署、A/B测试
---
## 算法简历要点
### ✅ DO
- 明确算法方向（NLP/CV/推荐/机器学习）
- 模型效果量化（准确率、召回率、F1、AUC等）
- 数据规模（训练数据量、特征维度）
- 模型复杂度（参数量、推理速度）
- 论文/竞赛/开源项目加分
### ❌ DON'T
- 只写"使用深度学习模型"不写明具体模型
- 不写模型效果和baseline对比
- 不写数据规模和场景
- 罗列算法名词但没有实际应用
---
## 示例对比
### ❌ 原写法
使用机器学习算法做用户流失预测，准确率还可以。
### ✅ 优化写法
模型AUC提升8% | 构建用户流失预测模型，精准识别高流失风险用户
- 数据处理：处理X万用户数据，构建X维特征（用户行为、交易、社交等）
- 模型设计：
- 对比LR、XGBoost、LightGBM，最终选用LightGBM
- 特征重要性分析，筛选Top 20核心特征
- 模型融合（5折交叉验证），AUC从0.72提升至0.80
- 业务应用：
- 识别高流失风险用户准确率85%，召回率75%
- 支撑运营精细化触达策略，用户留存率提升5%
- 模型每日自动更新，实时预测
FILE:assets/examples/杂乱信息整理示例.md
# 杂乱信息整理示例
## 示例1：时间混乱的信息
### ❌ 用户原始输入
"我做过一个春节活动效果挺好的，然后之前国庆也做过一个活动，还有平时在做数据分析。对了，还有一个和搜索相关的项目，也是做优化什么的。在字节待了半年左右吧，大概2023年底到2024年中。"
### ✅ 整理后
**识别到的时间线：**
- 2023.12 - 2024.06：字节跳动实习（约6个月）
**拆分出的核心项目（按时间倒序）：**
| 项目 | 时间 | 优先级 | 说明 |
|:---|:---|:---:|:---|
| 春节活动 | 2024.01-2024.02 | P0 | 有明确成果，重点展开 |
| 搜索优化项目 | 2024.02-2024.04？ | P1 | 需要确认具体时间 |
| 数据分析工作 | 贯穿实习期？ | P1/P2 | 需要确认是独立项目还是支持性工作 |
| 国庆活动 | 2023.09-2023.10 | P1 | 时间冲突！需要确认是否在字节前 |
**需要向用户确认的问题：**
1. 国庆活动是在字节做的，还是之前的经历？
2. 春节活动和搜索项目是并行的，还是有先后顺序？
3. 数据分析是某个项目的一部分，还是独立的日常工作？
---
## 示例2：多段经历混杂
### ❌ 用户原始输入
"我在B站做过用户增长，DAU提升了很多。然后也在腾讯待过，做产品经理，写过PRD文档，也做过数据分析，还参加过一些会议。后来去字节做策略产品，做了推荐策略，也做过一些增长活动，还有搜索相关的优化。对了，在学校还做过一个创业项目。"
### ✅ 整理后
**识别到的公司/组织：**
1. 字节跳动 - 策略产品
2. 腾讯 - 产品经理
3. B站 - 用户增长
4. 学校 - 创业项目
**按时间倒序排列：**
| 公司 | 岗位 | 识别到的项目 | 优先级 | 备注 |
|:---|:---|:---|:---:|:---|
| 字节 | 策略产品 | 推荐策略 | P0 | 核心项目 |
| 字节 | 策略产品 | 增长活动 | P1 | 重要项目 |
| 字节 | 策略产品 | 搜索优化 | P1 | 重要项目 |
| 腾讯 | 产品经理 | PRD撰写+需求落地 | P1 | 需要确认具体项目 |
| 腾讯 | 产品经理 | 数据分析 | P2 | 可能是支持性工作 |
| B站 | 用户增长 | 用户增长项目 | P0 | 有数据成果，重点展开 |
| 学校 | 创始人 | 创业项目 | P1 | 非实习经历，视情况保留 |
**过滤掉的信息：**
- ❌ "参加过一些会议" → 日常事务，删除
- ⚠️ "写过PRD文档" → 需要确认是哪个项目的一部分
---
## 示例3：大量无关信息
### ❌ 用户原始输入
"我每天早上9点上班，6点下班，中午休息1小时。我们组有5个人，leader人很好，经常请我们喝奶茶。我每天的工作包括写日报，每周要写周报，每月还有月报。我还帮同事订过外卖，组织过一次团建。在工作上，我做过一个AI项目，用ChatGPT做一些事情，效果还不错。还有一个数据分析的工作，给运营提供了一些支持。"
### ✅ 整理后
**识别到的核心信息：**
1. AI项目（使用ChatGPT）- P0 核心项目
2. 数据分析工作（支持运营）- P1 重要项目
**过滤掉的信息：**
| 信息 | 原因 |
|:---|:---|
| 上下班时间 | 与岗位能力无关 |
| 团队人数 | 非项目内容 |
| leader请奶茶 | 无关信息 |
| 日报/周报/月报 | 日常事务 |
| 订外卖、组织团建 | 无关信息 |
**需要向用户确认：**
1. AI项目具体做了什么？使用ChatGPT解决了什么问题？有什么成果？
2. 数据分析支持运营，具体分析了什么？输出了什么结论？有什么效果？
---
## 示例4：项目边界模糊
### ❌ 用户原始输入
"我在美团做商家运营，做了很多工作。包括商家的入驻审核，还有一些商家的培训，然后也做过一些活动让商家参与，还分析过商家的数据，给商家做过一些运营方案。有时候也处理商家的投诉和问题。"
### ✅ 整理后
**识别到的核心模块：**
| 模块 | 内容 | 优先级 | 处理方式 |
|:---|:---|:---:|:---|
| 商家入驻 | 审核、培训 | P1 | 合并为"商家入驻体系" |
| 商家活动 | 活动策划、商家参与 | P0 | 独立项目，重点展开 |
| 数据分析 | 商家数据分析 | P1 | 可以合并到活动或独立展示 |
| 运营方案 | 给商家的运营方案 | P2 | 可能是日常支持工作 |
| 投诉处理 | 客服/支持工作 | P3 | 删除或与岗位无关则简写 |
**关联合并建议：**
```
建议合并为2-3个核心项目：
项目1（P0）：商家增长活动策划
- 商家入驻审核与培训体系搭建
- 商家活动策划与执行
- 数据分析与效果评估
项目2（P1）：商家数据分析与运营支持
- 商家数据监控与分析
- 输出商家运营方案
- （投诉处理可删除或一句话带过）
```
---
## 示例5：多段实习经历混乱
### ❌ 用户原始输入
"第一段在小红书，做内容运营，负责创作者的对接，还有做一些活动。第二段在字节，做的是产品，不过也做了一些运营的工作，还有数据分析。然后之前还在腾讯实习过，也是做产品，做了一个增长的项目。对了，字节的时候还有一个AI相关的项目。"
### ✅ 整理后
**按时间倒序排列（假设）：**
| 时间 | 公司 | 岗位 | 项目 | 优先级 |
|:---|:---|:---|:---|:---:|
| 最近 | 字节 | 产品 | AI项目 | P0 |
| 最近 | 字节 | 产品 | 增长项目 | P0 |
| 最近 | 字节 | 产品 | 数据分析 | P1/P2 |
| 中间 | 小红书 | 内容运营 | 创作者运营 | P1 |
| 中间 | 小红书 | 内容运营 | 活动策划 | P1 |
| 最早 | 腾讯 | 产品 | 增长项目 | P0 |
**整理说明：**
1. 字节的产品岗位需要拆分3个项目
2. 小红书的内容运营可以合并为1-2个项目
3. 腾讯的增长项目与字节的合并展示或分别展示
**需要确认：**
1. 具体时间顺序
2. 字节的数据分析是独立项目还是其他项目的一部分
3. 字节的AI项目和增长项目是并行的还是有先后
---
## 整理原则总结
### 1. 去噪原则
- ✅ 保留：项目成果、职责、数据、技能
- ❌ 删除：日常事务、无关细节、主观感受
### 2. 拆分原则
- 一段实习经历 → 2-4个核心项目
- 每个项目有明确的边界和成果
### 3. 合并原则
- 同一业务的不同环节 → 合并展示
- 同一技术的不同应用 → 合并展示
- 支持性工作 → 合并到主项目或简写
### 4. 排序原则
- 按时间倒序（最近→最早）
- 同一公司内按重要性（P0→P1→P2）
### 5. 优先级标记
| 优先级 | 特征 | 简历处理方式 |
|:---|:---|:---|
| P0 | 有数据、独立负责、与岗位高度相关 | 重点展开，3-4个bullet point |
| P1 | 有参与、有学习价值 | 正常描述，2-3个bullet point |
| P2 | 参与支持、技能相关 | 简写，1-2个bullet point |
| P3 | 日常事务、无关 | 删除 |
FILE:assets/examples/示例输入.md
# 示例项目描述输入
## 示例1：简单文字描述
【用户输入】
我第一段实习在B站做产品实习生，主要负责AI搜索补词策略，替代了2-3名运营人力。第二段在得物做履约优化，降低了商家成本。
---
【系统提取后询问】
已识别到2段实习经历，需要补充以下信息：
**项目1：B站 - 产品实习生**
- 具体时间？（如2025.06-至今）
- 部门？（如OGV产品部）
- 具体成果数据？（如覆盖XX个IP，补词效率提升XX%）
- 使用的工具/技能？（如SQL、Prompt工程）
**项目2：得物 - 产品实习生**
- 具体时间？
- 部门？（如国际事业部）
- 商家成本降低多少？（如从280元降至80元）
- 具体负责什么模块？
---
## 示例2：较完整的文字描述
【用户输入】
实习经历：
1. B站 OGV产品部 产品实习生 2025.06-至今
- AI策略能力建设：设计AI搜索补词策略，替代2-3名运营人力/周
- 2025萌战：负责活动产品，用户覆盖143万，超预期186%
- 跨晚拜年纪：优化预约转化，转化率提升4.3%
2. 得物 国际事业部 产品实习生 2025.03-2025.06
- 新个卖限制出价：设计熔断机制，月均挽回4.2万损失
- 云鉴直发：推动直发流程，商家成本降低71%，GMV提升4.3%
教育背景：上海大学 管理科学与工程硕士 2024.09-2027.04
---
【系统提取结果】
已提取到以下信息：
**工作经历**
1. B站 - OGV产品部 - 产品实习生（2025.06-至今）
- 项目1：AI策略能力建设
- 项目2：2025萌战
- 项目3：跨晚&拜年纪
2. 得物 - 国际事业部 - 产品实习生（2025.03-2025.06）
- 项目1：新个卖限制出价
- 项目2：云鉴直发
**教育背景**
- 上海大学 - 管理科学与工程 - 硕士（2024.09-2027.04）
请确认：
1. 目标岗位方向？（产品经理/产品运营/数据分析/商业化/策略）
2. 求职类型？（校招/暑期/日常）
3. 是否需要补充更多细节？
---
## 示例3：飞书文档输入（模拟）
【飞书文档内容结构】
# 沈泽栋 - 实习经历总结
## B站 OGV产品部 产品实习生（2025.06-至今）
### 项目一：AI策略能力建设
**背景**：OGV的IP上线与宣发长期依赖运营基于经验手动完成搜索补词与投放策略拆解，存在漏配风险与重复劳动成本高的问题。
**我的职责**：
- 独立负责搜索补词和投放策略的AI化方案设计
- 将补词逻辑拆解为别名扩展、主创关联、错别字生成、同音词生成与组合词五类规则模块
- 打通规则建模→AI生成→校验→系统写入的工程闭环
**成果**：
- 替代2-3名运营人力/周
- 补词覆盖率提升XX%
- 投放策略一致性显著改善
### 项目二：2025萌战
**背景**：萌战为B站核心二次元角色票选活动，停办七年后重启。
**我的职责**：
- 负责提名机制、赛制结构、投票流程与分享模块功能落地
- 对活动组件进行结构化拆解以支持后续复用
- 活动期间持续跟踪投票数据与系统表现
**成果**：
- 活动期覆盖用户143w+，超预期186.2%
- 相关稿件VV 2200w，超预期20%
### 项目三：跨晚 & 拜年纪
**背景**：承接跨晚与拜年纪年度重点内容宣发，搭建"预约承接—互动参与"专题页能力。
**我的职责**：
- 统一主站与OTT预约数据口径，重构预约组件与数据合并逻辑
- 推动预约模块配置化与组件化
- 新增直播互动游戏提升预约转化率
**成果**：
- 跨晚总预约395w，同比增加2.3%
- 拜年纪总预约230w，同比增加1.3%
- 预约转化率增加4.3%
## 得物 国际事业部 产品实习生（2025.03-2025.06）
### 项目一：新个卖限制出价节点
**背景**：新卖家利用发货节点逻辑漏洞，在未真实到货情况下快速累积单量并逃避系统监管导致平台资损。
**我的职责**：
- 设计结算节点熔断机制
- 实施未真实完成首单结算的商家50单限额熔断
**成果**：
- 月均直接挽回4.2w
- 新卖家虚假发货率从2.7%降低至1.6%
- 6个月避免无效交易超预期32%
### 项目二：云鉴直发
**背景**：美国区存在常规业务需多段物流与质检，整体履约流程耗时长、成本高。
**我的职责**：
- 参与推动"商家拍照→云鉴审核→直发买家"流程
- 通过AI+人工复核的方式审核后，商家直接发货到买家
**成果**：
- 商家单位成本从280元降低至80元
- 物流效率提升30%
- 月GMV提升4.3%
---
## 示例4：含缺失信息的描述
【用户输入】
我在腾讯做了产品实习，主要负责一个增长项目，效果不错。还在字节待过一段时间，做数据分析。
---
【系统提取后询问】
识别到2段经历，但信息严重不足，需要补充：
**项目1：腾讯 - 产品实习**
- 具体时间？（起止年月）
- 部门/团队？
- 增长项目具体是什么？
- 你的具体职责？
- 有什么成果数据？
- 使用的工具/技能？
**项目2：字节 - 数据分析**
- 具体时间？
- 部门/团队？
- 具体分析了什么？
- 有什么发现和成果？
- 使用的工具（SQL/Python等）？
建议提供更详细的信息，或直接上传简历文件/文档链接。
---
## 示例5：多段零散经历
【用户输入】
1. 小红书 社区运营实习 2024.06-2024.09
- 负责创作者运营，对接100+创作者
- 策划创作者激励活动，参与创作者300+
- 优化创作者入驻流程，入驻率提升15%
2. 美团 产品运营实习 2024.03-2024.06
- 负责商家增长，新增商家500+
- 优化商家入驻流程，审核时效缩短30%
- 设计商家培训体系，商家满意度4.5/5
3. 校内创业 联合创始人 2023.09-2024.03
- 做校园二手交易平台
- 用户2000+，月GMV 5万
4. 学生会 宣传部长 2022.09-2023.06
- 负责公众号运营，粉丝从500增至3000
- 策划10+场活动，参与人数累计2000+
---
【系统提取后询问】
已提取到4段经历：
**实习经历（2段）**
1. 小红书 - 社区运营 - 2024.06-2024.09
2. 美团 - 产品运营 - 2024.03-2024.06
**其他经历（2段）**
3. 校内创业 - 联合创始人 - 2023.09-2024.03
4. 学生会 - 宣传部长 - 2022.09-2023.06
**需要确认：**
1. 目标岗位方向？（产品运营/内容运营/用户运营）
2. 求职类型？（校招/暑期实习/日常实习）
3. 其他经历是否需要保留？（建议校招保留创业，实习可不写学生会）
---
## 使用建议
**输入技巧：**
1. 时间、公司、岗位尽量写清楚
2. 有数据的尽量写数据
3. 没数据的写具体做了什么
4. 可以分段输入，系统会合并
**系统会做的：**
1. 自动提取关键信息
2. 识别缺失信息并询问
3. 按时间排序
4. 合并相似项目
5. 根据岗位方向优化表述