commit 175b38138efa75bcf30368dfaddc497f4b8375ab Author: 李清 Date: Tue Aug 4 14:02:45 2026 +0800 holy后端 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..1e78487 --- /dev/null +++ b/.env.example @@ -0,0 +1,10 @@ +HOST=0.0.0.0 +PORT=8000 +DATABASE_PATH=./data/holy_crab.db +FILE_STORAGE_PATH=./data/files +CRAWLER_BASE_URL=http://43.139.175.114:12007 +CRAWLER_RETRY_TIMES=3 +CRAWLER_RETRY_DELAY_SECONDS=2 +CRAWLER_TIMEOUT_SECONDS=120 +CRAWLER_POLL_INTERVAL_SECONDS=5 +CORS_ORIGIN=http://localhost:5173 diff --git a/.idea/Holy蟹搜索监测平台-后端服务-20260723.iml b/.idea/Holy蟹搜索监测平台-后端服务-20260723.iml new file mode 100644 index 0000000..3e3f564 --- /dev/null +++ b/.idea/Holy蟹搜索监测平台-后端服务-20260723.iml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..1452c1a --- /dev/null +++ b/README.md @@ -0,0 +1,43 @@ +# Holy蟹搜索监测平台业务后端 + +该服务位于 Web 前端与现有小红书爬虫服务之间： + +```text +Web 前端 -> 本业务后端（默认 8000）-> xiaoti-framework 爬虫服务（默认 5000） +``` + +业务后端负责生成公开任务 ID、保存任务列表、异步调用爬虫、同步状态与进度，以及保存结果文件。浏览器不再直接访问爬虫接口。 + +## 启动 + +```bash +cd "/Users/xiaoti/Downloads/Holy蟹搜索监测平台-后端服务-20260723" +python3 -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt +cp .env.example .env +python app.py +``` + +默认连接远程爬虫服务 `http://43.139.175.114:12007`。如地址不同，设置： + +```bash +export CRAWLER_BASE_URL="http://实际爬虫地址:端口" +``` + +SQLite 数据库默认保存在 `data/holy_crab.db`，任务文件保存在 `data/files/`。 + +## 已实现接口 + +- `POST /api/v1/tasks`：创建任务并立即返回业务任务 ID +- `GET /api/v1/tasks`：任务列表、搜索、状态筛选和分页 +- `GET /api/v1/tasks/{taskId}`：任务状态、进度、结果及文件 +- `POST /api/v1/tasks/{taskId}/cancel`：取消任务 +- `POST /api/v1/tasks/{taskId}/retry`：重新发起任务 +- `GET /api/v1/tasks/{taskId}/files`：文件列表 +- `GET /api/v1/tasks/{taskId}/files/{fileId}/preview`：预览 +- `GET /api/v1/tasks/{taskId}/files/{fileId}/download`：下载 + +当前接入规则与原前端保持一致：深度任务调用爬虫 `/api/v1/xhs/wen/sources`，轻度和常规任务调用 `/api/v1/xhs/wen/overview`。 + +取消操作会立即把业务任务标记为 `cancelled`，业务 Worker 不再接收爬虫结果。现有爬虫服务没有取消接口，因此已经发出的远端采集请求不能被强制中止；后续如爬虫服务增加取消接口，可在 `cancel_task` 中继续转发。 diff --git a/apifox-business-tasks.openapi.json b/apifox-business-tasks.openapi.json new file mode 100644 index 0000000..97c1f64 --- /dev/null +++ b/apifox-business-tasks.openapi.json @@ -0,0 +1,71 @@ +{ + "openapi": "3.0.3", + "info": { + "title": "Holy蟹业务任务接口", + "version": "1.1.0", + "description": "前端调用的业务后端任务接口。继续采集沿用原任务 ID；重新采集创建新任务。" + }, + "servers": [ + { + "url": "http://127.0.0.1:8000", + "description": "本地业务后端，请在 Apifox 环境中改成实际地址" + } + ], + "paths": { + "/api/v1/tasks/{task_id}/cancel": { + "post": { + "summary": "停止任务", + "tags": ["问一问任务控制"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "requestBody": { + "required": false, + "content": { + "application/json": { + "schema": {"type": "object", "properties": {"reason": {"type": "string"}}}, + "example": {"reason": "用户主动取消"} + } + } + }, + "responses": {"200": {"description": "停止请求成功"}} + } + }, + "/api/v1/tasks/{task_id}/retry": { + "post": { + "summary": "继续采集（兼容路径）", + "description": "沿用业务任务 ID 和爬虫任务 ID，从 framework 检查点继续。", + "tags": ["问一问任务控制"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": {"200": {"description": "恢复成功"}} + } + }, + "/api/v1/tasks/{task_id}/resume": { + "post": { + "summary": "继续采集", + "description": "沿用业务任务 ID 和爬虫任务 ID，从 framework 检查点继续。", + "tags": ["问一问任务控制"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": {"200": {"description": "恢复成功"}} + } + }, + "/api/v1/tasks/{task_id}/recollect": { + "post": { + "summary": "重新全量采集", + "description": "复制原任务配置，创建新的业务任务 ID 和爬虫任务 ID，从头执行。", + "tags": ["问一问任务控制"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": {"201": {"description": "新任务创建成功"}} + } + } + }, + "components": { + "parameters": { + "TaskId": { + "name": "task_id", + "in": "path", + "required": true, + "description": "Holy蟹业务任务 ID，例如 HC-20260723-ABCDE", + "schema": {"type": "string"} + } + } + } +} diff --git a/app.py b/app.py new file mode 100644 index 0000000..19d1af2 --- /dev/null +++ b/app.py @@ -0,0 +1,296 @@ +import secrets +from pathlib import Path +from urllib.parse import quote + +from flask import Flask, g, jsonify, redirect, request, send_file + +from auth_service import ( + build_dingtalk_login_url, + exchange_dingtalk_user, + get_user_by_session_token, + revoke_session, + upsert_user_and_session, +) +from config import Config +from database import init_database +from rule_service import ( + delete_rule_document, + get_rule_document, + get_rule_html_path, + save_rule_document, +) +from task_service import ( + TaskError, + cancel_task, + create_task, + get_file, + get_task, + get_task_extracted, + list_tasks, + recollect_task, + resume_task, + resume_unfinished_tasks, +) + + +def create_app() -> Flask: + app = Flask(__name__) + init_database() + + @app.before_request + def assign_request_id(): + g.request_id = request.headers.get("X-Request-Id") or f"req_{secrets.token_hex(10)}" + if request.method == "OPTIONS": + return None + open_paths = ("/health", "/api/v1/auth/", "/api/v1/skill/") + if request.path == "/health" or any(request.path.startswith(path) for path in open_paths): + return None + user = get_user_by_session_token(request.cookies.get(Config.SESSION_COOKIE_NAME)) + if user is None and request.path.startswith("/api/"): + return jsonify({ + "error": {"code": "UNAUTHORIZED", "message": "请先通过钉钉授权登录"}, + "request_id": g.request_id, + }), 401 + g.current_user = user + + @app.after_request + def add_headers(response): + response.headers["X-Request-Id"] = g.get("request_id", "") + response.headers["Access-Control-Allow-Origin"] = Config.CORS_ORIGIN + response.headers["Access-Control-Allow-Headers"] = "Content-Type,Authorization,Idempotency-Key,X-Request-Id" + response.headers["Access-Control-Allow-Methods"] = "GET,POST,DELETE,OPTIONS" + response.headers["Access-Control-Allow-Credentials"] = "true" + return response + + @app.errorhandler(TaskError) + def handle_task_error(error: TaskError): + return jsonify({ + "error": {"code": error.code, "message": error.message, "details": error.details}, + "request_id": g.request_id, + }), error.status_code + + @app.errorhandler(Exception) + def handle_unexpected(error: Exception): + app.logger.exception("未处理异常") + return jsonify({ + "error": {"code": "INTERNAL_ERROR", "message": "服务器内部错误"}, + "request_id": g.get("request_id"), + }), 500 + + @app.route("/health", methods=["GET"]) + def health(): + return jsonify({"status": "ok", "request_id": g.request_id}) + + @app.route("/api/v1/auth/dingtalk/login", methods=["GET"]) + def dingtalk_login(): + return jsonify({ + "data": build_dingtalk_login_url(request.args.get("redirect", "/tasks")), + "request_id": g.request_id, + }) + + @app.route("/api/v1/auth/dingtalk/callback", methods=["GET"]) + def dingtalk_callback(): + code = request.args.get("authCode") or request.args.get("code") + state = request.args.get("state", "") + frontend_redirect = "/tasks" + if ":" in state: + frontend_redirect = state.split(":", 1)[1] or frontend_redirect + if not code: + return redirect(f"{Config.FRONTEND_BASE_URL}/login?error={quote('钉钉授权缺少 code')}") + user_payload = exchange_dingtalk_user(code) + user, token, expires_at = upsert_user_and_session(user_payload) + response = redirect(f"{Config.FRONTEND_BASE_URL}{frontend_redirect}") + response.set_cookie( + Config.SESSION_COOKIE_NAME, + token, + expires=expires_at, + httponly=True, + secure=Config.SESSION_COOKIE_SECURE, + samesite=Config.SESSION_COOKIE_SAMESITE, + path="/", + ) + # 给前端“Link AI”复制提示词使用。后端鉴权仍读取 hc_session； + # 前端只读取这个辅助 Cookie 的 value，并在提示词中拼成 hc_session=。 + response.set_cookie( + "hc_session_agent", + token, + expires=expires_at, + httponly=False, + secure=Config.SESSION_COOKIE_SECURE, + samesite=Config.SESSION_COOKIE_SAMESITE, + path="/", + ) + app.logger.info("钉钉用户登录成功 user_id=%s name=%s", user["id"], user["name"]) + return response + + @app.route("/api/v1/auth/direct-token", methods=["GET"]) + def auth_direct_token(): + """直接用钉钉 authCode 换取 session token，AI 无需浏览器 Cookie。 + + 调用方式： + GET /api/v1/auth/direct-token?authCode=钉钉返回的code + + 返回： + {"data": {"token": "session_token", "user": {...}, "expires_at": "..."}} + + AI 拿到 token 后，后续请求带上： + Cookie: hc_session= + """ + code = request.args.get("authCode") or request.args.get("code") + if not code: + return jsonify({"code": "MISSING_AUTH_CODE", "msg": "缺少 authCode 参数"}), 400 + try: + dingtalk_user = exchange_dingtalk_user(code) + user, token, expires_at_dt = upsert_user_and_session(dingtalk_user) + return jsonify({ + "data": { + "token": token, + "user": user, + "expires_at": expires_at_dt.isoformat(timespec="seconds"), + }, + "request_id": g.request_id, + }) + except TaskError as e: + return jsonify({"code": e.code, "msg": e.message, "data": e.data}), e.http_status + + # ── Skill 版本管理 ──────────────────────────────────────────── + SKILL_BASE = Path(__file__).parent / "skills" + + @app.route("/api/v1/skill/holy-crab/version", methods=["GET"]) + def skill_holy_crab_version(): + """返回当前 Holy Crab Skill 的版本信息。""" + import json + skill_dir = SKILL_BASE / "holy-crab" + version_file = skill_dir / "VERSION" + if not version_file.exists(): + return jsonify({"code": "SKILL_NOT_FOUND", "msg": "Skill 文件不存在"}), 404 + info = json.loads(version_file.read_text(encoding="utf-8")) + return jsonify({"data": {"version": info}, "request_id": g.request_id}) + + @app.route("/api/v1/skill/holy-crab/download", methods=["GET"]) + def skill_holy_crab_download(): + """下载 Holy Crab Skill 完整 zip 包（含所有文件）。""" + import hashlib, io, zipfile + skill_dir = SKILL_BASE / "holy-crab" + if not skill_dir.exists(): + return jsonify({"code": "SKILL_NOT_FOUND", "msg": "Skill 文件不存在"}), 404 + + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + for fpath in skill_dir.rglob("*"): + if fpath.is_file(): + arcname = str(fpath.relative_to(skill_dir)) + zf.writestr(arcname, fpath.read_bytes()) + buf.seek(0) + return buf.getvalue(), 200, { + "Content-Type": "application/zip", + "Content-Disposition": "attachment; filename=holy-crab-skill.zip", + "X-Skill-Version": json.loads((skill_dir / "VERSION").read_text(encoding="utf-8")).get("version", "unknown"), + } + + @app.route("/api/v1/auth/me", methods=["GET"]) + def auth_me(): + user = get_user_by_session_token(request.cookies.get(Config.SESSION_COOKIE_NAME)) + if not user: + raise TaskError("UNAUTHORIZED", "请先通过钉钉授权登录", 401) + return jsonify({"data": {"user": user}, "request_id": g.request_id}) + + @app.route("/api/v1/auth/logout", methods=["POST"]) + def auth_logout(): + revoke_session(request.cookies.get(Config.SESSION_COOKIE_NAME)) + response = jsonify({"data": {"ok": True}, "request_id": g.request_id}) + response.delete_cookie(Config.SESSION_COOKIE_NAME, path="/") + response.delete_cookie("hc_session_agent", path="/") + return response + + @app.route("/api/v1/tasks", methods=["POST"]) + def create_research_task(): + payload = request.get_json(silent=True) or {} + current_user = getattr(g, "current_user", None) or {} + payload["created_by_id"] = current_user.get("id") or payload.get("created_by_id") + payload["created_by_name"] = current_user.get("name") or payload.get("created_by_name") + task = create_task(payload) + return jsonify({"data": task, "request_id": g.request_id}), 201 + + @app.route("/api/v1/tasks", methods=["GET"]) + def query_research_tasks(): + result = list_tasks( + q=request.args.get("q", "").strip(), + status=request.args.get("status", "").strip(), + page=int(request.args.get("page", "1")), + page_size=int(request.args.get("page_size", "10")), + ) + return jsonify({"data": result, "request_id": g.request_id}) + + @app.route("/api/v1/tasks/", methods=["GET"]) + def get_research_task(task_id: str): + if request.args.get("view") in {"extracted", "cleaned", "parsed"}: + return jsonify({"data": get_task_extracted(task_id), "request_id": g.request_id}) + return jsonify({"data": get_task(task_id), "request_id": g.request_id}) + + @app.route("/api/v1/tasks//extracted", methods=["GET"]) + def get_research_task_extracted(task_id: str): + return jsonify({"data": get_task_extracted(task_id), "request_id": g.request_id}) + + @app.route("/api/v1/tasks//cancel", methods=["POST"]) + def cancel_research_task(task_id: str): + body = request.get_json(silent=True) or {} + return jsonify({ + "data": cancel_task(task_id, str(body.get("reason") or "")), + "request_id": g.request_id, + }) + + @app.route("/api/v1/tasks//retry", methods=["POST"]) + @app.route("/api/v1/tasks//resume", methods=["POST"]) + def resume_research_task(task_id: str): + return jsonify({"data": resume_task(task_id), "request_id": g.request_id}) + + @app.route("/api/v1/tasks//recollect", methods=["POST"]) + def recollect_research_task(task_id: str): + return jsonify({ + "data": recollect_task(task_id), + "request_id": g.request_id, + }), 201 + + @app.route("/api/v1/tasks//files", methods=["GET"]) + def list_task_files(task_id: str): + task = get_task(task_id) + return jsonify({"data": {"items": task["files"]}, "request_id": g.request_id}) + + @app.route("/api/v1/tasks//files//preview", methods=["GET"]) + def preview_task_file(task_id: str, file_id: str): + metadata, path = get_file(task_id, file_id) + if not metadata["previewable"]: + raise TaskError("FILE_TOO_LARGE", "文件超过预览大小限制", 413) + return send_file(path, as_attachment=False) + + @app.route("/api/v1/tasks//files//download", methods=["GET"]) + def download_task_file(task_id: str, file_id: str): + _, path = get_file(task_id, file_id) + return send_file(path, as_attachment=True, download_name=Path(path).name) + + @app.route("/api/v1/rules", methods=["GET"]) + def get_current_rule(): + return jsonify({"data": get_rule_document(), "request_id": g.request_id}) + + @app.route("/api/v1/rules/current.html", methods=["GET"]) + def preview_current_rule_html(): + return send_file(get_rule_html_path(), mimetype="text/html; charset=utf-8") + + @app.route("/api/v1/rules/html", methods=["POST"]) + def upload_rule_html(): + return jsonify({"data": save_rule_document(request), "request_id": g.request_id}) + + @app.route("/api/v1/rules/html", methods=["DELETE"]) + def remove_rule_html(): + return jsonify({"data": delete_rule_document(), "request_id": g.request_id}) + + resume_unfinished_tasks() + return app + + +app = create_app() + + +if __name__ == "__main__": + app.run(host=Config.HOST, port=Config.PORT, threaded=True) diff --git a/auth_service.py b/auth_service.py new file mode 100644 index 0000000..7ee6473 --- /dev/null +++ b/auth_service.py @@ -0,0 +1,171 @@ +import hashlib +import secrets +from datetime import datetime, timedelta +from typing import Any +from urllib.parse import urlencode + +import requests + +from config import Config +from database import connection, json_dumps, json_loads, now_iso +from task_service import TaskError + + +def _sha256(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _normalize_user(payload: dict[str, Any]) -> dict[str, str]: + user_id = ( + payload.get("unionId") + or payload.get("unionid") + or payload.get("userid") + or payload.get("userId") + or payload.get("openId") + or payload.get("openid") + ) + name = ( + payload.get("nick") + or payload.get("nickName") + or payload.get("name") + or payload.get("mobile") + or user_id + ) + if not user_id: + raise TaskError("DINGTALK_USER_ID_MISSING", "钉钉用户信息中缺少用户标识", 502, {"payload": payload}) + return { + "ding_user_id": str(user_id), + "union_id": str(payload.get("unionId") or payload.get("unionid") or ""), + "open_id": str(payload.get("openId") or payload.get("openid") or ""), + "name": str(name or ""), + "avatar": str(payload.get("avatarUrl") or payload.get("avatar") or ""), + "mobile": str(payload.get("mobile") or ""), + "email": str(payload.get("email") or ""), + } + + +def build_dingtalk_login_url(frontend_redirect: str = "/tasks") -> dict[str, str]: + if not Config.DINGTALK_CLIENT_ID: + raise TaskError("DINGTALK_NOT_CONFIGURED", "未配置钉钉 DINGTALK_CLIENT_ID/DINGTALK_APP_KEY", 500) + state = secrets.token_urlsafe(24) + params = { + "redirect_uri": Config.DINGTALK_REDIRECT_URI, + "response_type": "code", + "client_id": Config.DINGTALK_CLIENT_ID, + "scope": "openid", + "state": f"{state}:{frontend_redirect or '/tasks'}", + "prompt": "consent", + } + return {"url": f"{Config.DINGTALK_AUTH_URL}?{urlencode(params)}", "state": params["state"]} + + +def exchange_dingtalk_user(code: str) -> dict[str, Any]: + if not Config.DINGTALK_CLIENT_ID or not Config.DINGTALK_CLIENT_SECRET: + raise TaskError("DINGTALK_NOT_CONFIGURED", "未配置钉钉应用 client_id/client_secret", 500) + token_resp = requests.post( + Config.DINGTALK_TOKEN_URL, + json={ + "clientId": Config.DINGTALK_CLIENT_ID, + "clientSecret": Config.DINGTALK_CLIENT_SECRET, + "code": code, + "grantType": "authorization_code", + }, + timeout=20, + ) + if token_resp.status_code >= 400: + raise TaskError("DINGTALK_TOKEN_FAILED", "钉钉 access_token 获取失败", 502, {"body": token_resp.text}) + token_body = token_resp.json() + access_token = token_body.get("accessToken") or token_body.get("access_token") + if not access_token: + raise TaskError("DINGTALK_TOKEN_MISSING", "钉钉响应中缺少 accessToken", 502, {"body": token_body}) + user_resp = requests.get( + Config.DINGTALK_USER_URL, + headers={"x-acs-dingtalk-access-token": access_token}, + timeout=20, + ) + if user_resp.status_code >= 400: + raise TaskError("DINGTALK_USER_FAILED", "钉钉用户信息获取失败", 502, {"body": user_resp.text}) + return user_resp.json() + + +def upsert_user_and_session(dingtalk_user: dict[str, Any]) -> tuple[dict[str, Any], str, datetime]: + normalized = _normalize_user(dingtalk_user) + token = secrets.token_urlsafe(36) + session_hash = _sha256(token) + now = now_iso() + expires_at_dt = datetime.now().astimezone() + timedelta(days=Config.SESSION_EXPIRE_DAYS) + expires_at = expires_at_dt.isoformat(timespec="seconds") + with connection() as conn: + existing = conn.execute( + "SELECT id FROM auth_users WHERE ding_user_id=?", + (normalized["ding_user_id"],), + ).fetchone() + if existing: + user_id = existing["id"] + conn.execute( + """UPDATE auth_users SET union_id=?,open_id=?,name=?,avatar=?,mobile=?,email=?, + raw_json=?,last_login_at=?,updated_at=? WHERE id=?""", + ( + normalized["union_id"], normalized["open_id"], normalized["name"], + normalized["avatar"], normalized["mobile"], normalized["email"], + json_dumps(dingtalk_user), now, now, user_id, + ), + ) + else: + cursor = conn.execute( + """INSERT INTO auth_users + (ding_user_id,union_id,open_id,name,avatar,mobile,email,raw_json,last_login_at,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + ( + normalized["ding_user_id"], normalized["union_id"], normalized["open_id"], + normalized["name"], normalized["avatar"], normalized["mobile"], + normalized["email"], json_dumps(dingtalk_user), now, now, now, + ), + ) + user_id = cursor.lastrowid + conn.execute( + """INSERT INTO auth_sessions (session_hash,user_id,expires_at,created_at,last_seen_at) + VALUES (?,?,?,?,?)""", + (session_hash, user_id, expires_at, now, now), + ) + return get_user_by_session_token(token), token, expires_at_dt + + +def get_user_by_session_token(token: str | None) -> dict[str, Any] | None: + if not token: + return None + session_hash = _sha256(token) + current = now_iso() + with connection() as conn: + row = conn.execute( + """SELECT s.id AS session_id,s.expires_at,u.* FROM auth_sessions s + JOIN auth_users u ON u.id=s.user_id + WHERE s.session_hash=? AND s.revoked_at IS NULL AND s.expires_at>?""", + (session_hash, current), + ).fetchone() + if not row: + return None + conn.execute( + "UPDATE auth_sessions SET last_seen_at=? WHERE id=?", + (current, row["session_id"]), + ) + raw = json_loads(row["raw_json"]) if row["raw_json"] else {} + return { + "id": str(row["ding_user_id"]), + "name": row["name"], + "account": row["mobile"] or row["email"] or row["ding_user_id"], + "avatar": row["avatar"], + "union_id": row["union_id"], + "open_id": row["open_id"], + "raw": raw, + } + + +def revoke_session(token: str | None) -> None: + if not token: + return + with connection() as conn: + conn.execute( + "UPDATE auth_sessions SET revoked_at=? WHERE session_hash=? AND revoked_at IS NULL", + (now_iso(), _sha256(token)), + ) diff --git a/config.py b/config.py new file mode 100644 index 0000000..c598330 --- /dev/null +++ b/config.py @@ -0,0 +1,40 @@ +import os +from pathlib import Path + +from dotenv import load_dotenv + + +BASE_DIR = Path(__file__).resolve().parent +load_dotenv(BASE_DIR / ".env") + + +class Config: + """业务后端配置，生产环境应通过环境变量注入。""" + + HOST = os.getenv("HOST", "0.0.0.0") + PORT = int(os.getenv("PORT", "8000")) + DATABASE_PATH = Path(os.getenv("DATABASE_PATH", BASE_DIR / "data" / "holy_crab.db")) + FILE_STORAGE_PATH = Path(os.getenv("FILE_STORAGE_PATH", BASE_DIR / "data" / "files")) + CRAWLER_BASE_URL = os.getenv( + "CRAWLER_BASE_URL", + "http://43.139.175.114:12007", + ).rstrip("/") + CRAWLER_TIMEOUT_SECONDS = int(os.getenv("CRAWLER_TIMEOUT_SECONDS", "120")) + CRAWLER_RETRY_TIMES = int(os.getenv("CRAWLER_RETRY_TIMES", "3")) + CRAWLER_RETRY_DELAY_SECONDS = float(os.getenv("CRAWLER_RETRY_DELAY_SECONDS", "2")) + CRAWLER_POLL_INTERVAL_SECONDS = float(os.getenv("CRAWLER_POLL_INTERVAL_SECONDS", "5")) + CORS_ORIGIN = os.getenv("CORS_ORIGIN", "http://localhost:5173") + FRONTEND_BASE_URL = os.getenv("FRONTEND_BASE_URL", CORS_ORIGIN).rstrip("/") + SESSION_COOKIE_NAME = os.getenv("SESSION_COOKIE_NAME", "hc_session") + SESSION_COOKIE_SECURE = os.getenv("SESSION_COOKIE_SECURE", "true").lower() in ("1", "true", "yes", "on") + SESSION_COOKIE_SAMESITE = os.getenv("SESSION_COOKIE_SAMESITE", "Lax") + SESSION_EXPIRE_DAYS = int(os.getenv("SESSION_EXPIRE_DAYS", "14")) + DINGTALK_CLIENT_ID = os.getenv("DINGTALK_CLIENT_ID", os.getenv("DINGTALK_APP_KEY", "")) + DINGTALK_CLIENT_SECRET = os.getenv("DINGTALK_CLIENT_SECRET", os.getenv("DINGTALK_APP_SECRET", "")) + DINGTALK_REDIRECT_URI = os.getenv( + "DINGTALK_REDIRECT_URI", + f"{FRONTEND_BASE_URL}/api/v1/auth/dingtalk/callback", + ) + DINGTALK_AUTH_URL = os.getenv("DINGTALK_AUTH_URL", "https://login.dingtalk.com/oauth2/auth") + DINGTALK_TOKEN_URL = os.getenv("DINGTALK_TOKEN_URL", "https://api.dingtalk.com/v1.0/oauth2/userAccessToken") + DINGTALK_USER_URL = os.getenv("DINGTALK_USER_URL", "https://api.dingtalk.com/v1.0/contact/users/me") diff --git a/crawler_client.py b/crawler_client.py new file mode 100644 index 0000000..07c628a --- /dev/null +++ b/crawler_client.py @@ -0,0 +1,115 @@ +import re +import time +from typing import Any + +import requests + +from config import Config + + +class CrawlerError(RuntimeError): + pass + + +CRAWLER_SESSION = requests.Session() +# 爬虫服务为明确配置的直连地址，不应继承 macOS 系统代理。 +# 否则 requests 会经由 127.0.0.1:7890，约 30 秒后被代理返回 502。 +CRAWLER_SESSION.trust_env = False + + +def normalize_collection_keywords(keywords: list[str]) -> list[str]: + """清理影响小红书问一问卡片匹配的末尾标点，并保持关键词顺序。""" + + normalized: list[str] = [] + for keyword in keywords: + value = re.sub(r"[?？!！。．.]+$", "", str(keyword).strip()).strip() + if value and value not in normalized: + normalized.append(value) + if not normalized: + raise CrawlerError("关键词不能为空") + return normalized + + +def _post( + path: str, + body: dict[str, Any], + *, + retry_transient: bool = True, +) -> dict[str, Any]: + url = f"{Config.CRAWLER_BASE_URL}{path}" + retry_times = max(1, Config.CRAWLER_RETRY_TIMES) if retry_transient else 1 + response: requests.Response | None = None + + for attempt in range(1, retry_times + 1): + try: + response = CRAWLER_SESSION.post( + url, + json=body, + timeout=Config.CRAWLER_TIMEOUT_SECONDS, + ) + # 网关临时异常时自动重试，避免直接将业务任务标记为失败。 + if response.status_code in {502, 503, 504} and attempt < retry_times: + time.sleep(Config.CRAWLER_RETRY_DELAY_SECONDS * attempt) + continue + response.raise_for_status() + payload = response.json() + break + except (requests.ConnectionError, requests.Timeout) as exc: + if attempt < retry_times: + time.sleep(Config.CRAWLER_RETRY_DELAY_SECONDS * attempt) + continue + raise CrawlerError(f"爬虫服务请求失败（已重试 {retry_times} 次）：{exc}") from exc + except requests.RequestException as exc: + detail = response.text[:500] if response is not None else "" + message = f"爬虫服务请求失败：{exc}" + if detail: + message += f"，响应：{detail}" + raise CrawlerError(message) from exc + except ValueError as exc: + raise CrawlerError("爬虫服务返回的不是有效 JSON") from exc + else: + raise CrawlerError(f"爬虫服务请求失败（已重试 {retry_times} 次）") + + if payload.get("code") != 200 or not payload.get("success", False): + raise CrawlerError(payload.get("msg") or "爬虫服务返回失败") + data = payload.get("data") + if not isinstance(data, dict): + raise CrawlerError("爬虫服务响应缺少 data") + return data + + +def start_collection(mode: str, keywords: list[str]) -> dict[str, Any]: + """深度任务走 sources，轻度/常规任务走 overview。""" + + collection_keywords = normalize_collection_keywords(keywords) + path = "/api/v1/xhs/wen/sources" if mode == "deep" else "/api/v1/xhs/wen/overview" + # 远程 overview 只有接收数组时才创建异步任务并立即返回 task_id。 + # 因此轻度任务即使只有一个关键词也必须传数组，避免单关键词走同步 + # overview 并因采集耗时超过网关时限而返回 502。 + keyword: str | list[str] = ( + collection_keywords if mode == "mild" else collection_keywords[0] + ) + # 轻度 overview 在首次计算超过网关时限后，远程通常已经写好缓存， + # 允许受控重试以取得结果；深度 sources 会启动长任务，禁止盲目重试， + # 避免远程请求虽断开但后台继续执行时产生重复采集任务。 + return _post( + path, + {"keyword": keyword}, + retry_transient=(mode != "deep"), + ) + + +def query_collection(crawler_task_id: str) -> dict[str, Any]: + return _post("/api/v1/xhs/wen/tasks/query", {"task_id": crawler_task_id}) + + +def cancel_collection(crawler_task_id: str) -> dict[str, Any]: + """通知爬虫任务在下一个安全检查点停止并保存部分结果。""" + + return _post(f"/api/v1/xhs/wen/tasks/{crawler_task_id}/cancel", {}) + + +def resume_collection(crawler_task_id: str) -> dict[str, Any]: + """沿用原爬虫任务 ID，从 framework 保存的检查点继续采集。""" + + return _post(f"/api/v1/xhs/wen/tasks/{crawler_task_id}/resume", {}) diff --git a/database.py b/database.py new file mode 100644 index 0000000..bde8493 --- /dev/null +++ b/database.py @@ -0,0 +1,147 @@ +import json +import sqlite3 +from contextlib import contextmanager +from datetime import datetime +from typing import Any, Iterator + +from config import Config + + +def now_iso() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +@contextmanager +def connection() -> Iterator[sqlite3.Connection]: + conn = sqlite3.connect(Config.DATABASE_PATH, timeout=30) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + try: + yield conn + conn.commit() + finally: + conn.close() + + +def init_database() -> None: + """仅在服务启动时创建表，接口调用过程中不会重复执行建表。""" + + Config.DATABASE_PATH.parent.mkdir(parents=True, exist_ok=True) + Config.FILE_STORAGE_PATH.mkdir(parents=True, exist_ok=True) + with connection() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS research_tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + public_task_id TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + platform TEXT NOT NULL, + mode TEXT NOT NULL, + status TEXT NOT NULL, + execution_stage TEXT NOT NULL DEFAULT 'queued', + progress_percent INTEGER NOT NULL DEFAULT 0 CHECK(progress_percent BETWEEN 0 AND 100), + created_by_id TEXT NOT NULL, + created_by_name TEXT NOT NULL, + retry_of_task_id INTEGER, + crawler_task_id TEXT, + error_code TEXT, + error_message TEXT, + result_json TEXT, + cancel_requested_at TEXT, + started_at TEXT, + completed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + version INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY(retry_of_task_id) REFERENCES research_tasks(id) + ); + CREATE INDEX IF NOT EXISTS idx_tasks_status_created + ON research_tasks(status, created_at DESC); + CREATE INDEX IF NOT EXISTS idx_tasks_creator_created + ON research_tasks(created_by_id, created_at DESC); + + CREATE TABLE IF NOT EXISTS task_keywords ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + keyword_order INTEGER NOT NULL, + keyword TEXT NOT NULL, + normalized_keyword TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(task_id, keyword_order), + UNIQUE(task_id, normalized_keyword), + FOREIGN KEY(task_id) REFERENCES research_tasks(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + event_type TEXT NOT NULL, + stage TEXT, + progress_percent INTEGER, + actor_type TEXT NOT NULL, + actor_id TEXT, + payload_json TEXT, + occurred_at TEXT NOT NULL, + FOREIGN KEY(task_id) REFERENCES research_tasks(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS task_files ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + public_file_id TEXT NOT NULL UNIQUE, + task_id INTEGER NOT NULL, + name TEXT NOT NULL, + stage TEXT NOT NULL, + category TEXT NOT NULL, + format TEXT NOT NULL, + record_count INTEGER NOT NULL DEFAULT 0, + size_bytes INTEGER NOT NULL DEFAULT 0, + storage_key TEXT NOT NULL, + sha256 TEXT NOT NULL, + schema_version TEXT NOT NULL DEFAULT '1.0', + generated_at TEXT NOT NULL, + available INTEGER NOT NULL DEFAULT 1, + UNIQUE(task_id, name), + FOREIGN KEY(task_id) REFERENCES research_tasks(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS auth_users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ding_user_id TEXT NOT NULL UNIQUE, + union_id TEXT, + open_id TEXT, + name TEXT NOT NULL, + avatar TEXT, + mobile TEXT, + email TEXT, + raw_json TEXT, + last_login_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_auth_users_name + ON auth_users(name); + + CREATE TABLE IF NOT EXISTS auth_sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_hash TEXT NOT NULL UNIQUE, + user_id INTEGER NOT NULL, + expires_at TEXT NOT NULL, + created_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + revoked_at TEXT, + FOREIGN KEY(user_id) REFERENCES auth_users(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_auth_sessions_user + ON auth_sessions(user_id, expires_at); + """ + ) + + +def json_dumps(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +def json_loads(value: str | None) -> Any: + if not value: + return None + return json.loads(value) diff --git a/deploy/holy-crab.env b/deploy/holy-crab.env new file mode 100644 index 0000000..be8e835 --- /dev/null +++ b/deploy/holy-crab.env @@ -0,0 +1,18 @@ +HOST=127.0.0.1 +PORT=18000 +DATABASE_PATH=/var/lib/holy-crab/holy_crab.db +FILE_STORAGE_PATH=/var/lib/holy-crab/files +CRAWLER_BASE_URL=http://43.139.175.114:12007 +CRAWLER_RETRY_TIMES=3 +CRAWLER_RETRY_DELAY_SECONDS=2 +CRAWLER_TIMEOUT_SECONDS=120 +CRAWLER_POLL_INTERVAL_SECONDS=5 +CORS_ORIGIN=https://content.gbotai.cn +FRONTEND_BASE_URL=https://content.gbotai.cn/holy-crab +SESSION_COOKIE_NAME=hc_session +SESSION_COOKIE_SECURE=true +SESSION_COOKIE_SAMESITE=Lax +SESSION_EXPIRE_DAYS=14 +DINGTALK_CLIENT_ID=ding1a03sq3htguwcigp +DINGTALK_CLIENT_SECRET=1GhCT3t2iW2iqZ7FTEywDnR7u-aCMzCu1Kum6VGHGeNMk7hj9TIs2dsLy5vi3XXk +DINGTALK_REDIRECT_URI=https://content.gbotai.cn/holy-crab/api/v1/auth/dingtalk/callback diff --git a/deploy/holy-crab.service b/deploy/holy-crab.service new file mode 100644 index 0000000..3bd8b25 --- /dev/null +++ b/deploy/holy-crab.service @@ -0,0 +1,18 @@ +[Unit] +Description=Holy Crab Monitoring Platform Backend +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=ubuntu +Group=ubuntu +WorkingDirectory=/opt/holy-crab/backend +EnvironmentFile=/etc/holy-crab.env +ExecStart=/opt/holy-crab/backend/.venv/bin/gunicorn --bind 127.0.0.1:18000 --workers 1 --threads 8 --timeout 180 --access-logfile - --error-logfile - app:app +Restart=always +RestartSec=3 +TimeoutStopSec=30 + +[Install] +WantedBy=multi-user.target diff --git a/deploy/nginx-holy-crab.conf b/deploy/nginx-holy-crab.conf new file mode 100644 index 0000000..27539a2 --- /dev/null +++ b/deploy/nginx-holy-crab.conf @@ -0,0 +1,26 @@ +location = /holy-crab { + return 301 /holy-crab/; +} + +location ^~ /holy-crab/api/ { + proxy_pass http://127.0.0.1:18000/api/; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 180s; + proxy_send_timeout 180s; +} + +location = /holy-crab/health { + proxy_pass http://127.0.0.1:18000/health; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; +} + +location ^~ /holy-crab/ { + alias /opt/holy-crab/frontend/; + try_files $uri $uri/ /holy-crab/index.html; +} diff --git a/extract_service.py b/extract_service.py new file mode 100644 index 0000000..a362ec3 --- /dev/null +++ b/extract_service.py @@ -0,0 +1,470 @@ +""" +extract_service.py +================= +将爬虫原始响应（/api/v1/xhs/wen/overview · /api/v1/xhs/wen/sources） +清洗为标准结构化格式，与 extract_wenyiwen.py 逻辑一致。 + +在业务后端层面做此转换的好处： + - 任务结果存入 DB 时已是最干净的格式 + - 前端查看文件、AI 读取数据均直接拿到结构化结果 + - 无需前端再依赖原始爬虫字段名 +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Any + + +# --------------------------------------------------------------------------- +# 辅助：类型安全取值 +# --------------------------------------------------------------------------- + +def as_dict(value: Any, path: str, warnings: list[str]) -> dict[str, Any]: + if isinstance(value, dict): + return value + warnings.append(f"{path} 缺失或不是对象") + return {} + + +def as_list(value: Any, path: str, warnings: list[str]) -> list[Any]: + if isinstance(value, list): + return value + warnings.append(f"{path} 缺失或不是数组") + return [] + + +def field(item: dict[str, Any], key: str, path: str, warnings: list[str]) -> Any: + if key not in item: + warnings.append(f"{path}.{key} 缺失") + return None + return item[key] + + +# --------------------------------------------------------------------------- +# 辅助：数值解析 +# --------------------------------------------------------------------------- + +def first_integer(value: Any) -> int | None: + if not isinstance(value, str): + return None + match = re.search(r"\d[\d,]*", value) + if not match: + return None + return int(match.group(0).replace(",", "")) + + +def numeric_sort_value(value: Any) -> float | None: + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + parsed = first_integer(value) + return float(parsed) if parsed is not None else None + + +# --------------------------------------------------------------------------- +# 辅助：时间 / 视频时长格式化 +# --------------------------------------------------------------------------- + +def format_publish_time( + value: Any, path: str, warnings: list[str] +) -> str | None: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or value <= 0 + ): + if value != 0: + warnings.append(f"{path}.time 缺失或不是有效 Unix 时间戳") + return None + try: + china_tz = timezone(timedelta(hours=8)) + return datetime.fromtimestamp(value, china_tz).strftime("%Y-%m-%d %H:%M:%S") + except (OverflowError, OSError, ValueError): + warnings.append(f"{path}.time 超出可转换范围") + return None + + +def format_video_duration( + value: Any, path: str, warnings: list[str] +) -> str | None: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or value < 0 + ): + warnings.append(f"{path}.duration 缺失或不是非负数") + return None + total_seconds = int(value / 1000) if value >= 1000 else int(value) + hours, remainder = divmod(total_seconds, 3600) + minutes, seconds = divmod(remainder, 60) + if hours: + return f"{hours:02d}:{minutes:02d}:{seconds:02d}" + return f"{minutes:02d}:{seconds:02d}" + + +# --------------------------------------------------------------------------- +# 辅助：正文清洗 +# --------------------------------------------------------------------------- + +def clean_full_body(desc: Any, title: Any) -> str | None: + if not isinstance(desc, str): + return None + body = desc.replace("\r\n", "\n").replace("\r", "\n") + stripped_body = body.lstrip() + if isinstance(title, str) and title and stripped_body.startswith(title): + remainder = stripped_body[len(title):] + if not remainder or remainder[0].isspace(): + body = remainder.lstrip() + topic_pattern = r"#[^#\n]*?\[话题\]#?" + body = re.sub(rf"^(?:\s*{topic_pattern})+\s*", "", body) + trailing_topic = re.search(topic_pattern, body) + if trailing_topic: + topic_start = trailing_topic.start() + prior_topic_block = re.search( + r"(?s)(?:\s+#[^\n]*)+\s*$", body[:topic_start] + ) + if prior_topic_block: + topic_start = prior_topic_block.start() + else: + while topic_start > 0 and body[topic_start - 1] in " \t": + topic_start -= 1 + body = body[:topic_start] + else: + body = re.sub(r"(?s)#[^\n]*(?:\n[ \t]*#[^\n]*)*\s*$", "", body) + body = re.sub(r"(?m)^[ \t]*#+[ \t]*$", "", body) + body = re.sub(r"(?m)(?:^|\n)[ \t]*(?:标签|话题)[：:][ \t]*$", "", body) + return body.strip() + + +def body_char_count(body: str | None) -> int | None: + if body is None: + return None + return len(re.sub(r"\s+", "", body)) + + +# --------------------------------------------------------------------------- +# 召回明细：建立 (product_name, tag_name) → notes 的查找表 +# --------------------------------------------------------------------------- + +def build_recall_lookup( + core_data: dict[str, Any], base_path: str, warnings: list[str] +) -> tuple[dict[tuple[str, str], dict[str, Any]], bool]: + raw_note_details = core_data.get("note_details") + if raw_note_details is None: + return {}, False + note_details = as_dict(raw_note_details, f"{base_path}.note_details", warnings) + products = as_list( + note_details.get("products"), + f"{base_path}.note_details.products", + warnings, + ) + lookup: dict[tuple[str, str], dict[str, Any]] = {} + for product_index, raw_product in enumerate(products): + product_path = f"{base_path}.note_details.products[{product_index}]" + product = as_dict(raw_product, product_path, warnings) + product_name = field(product, "product_name", product_path, warnings) + content_tags = as_list( + product.get("content_tags"), + f"{product_path}.content_tags", + warnings, + ) + for tag_index, raw_tag in enumerate(content_tags): + tag_path = f"{product_path}.content_tags[{tag_index}]" + tag = as_dict(raw_tag, tag_path, warnings) + tag_name = field(tag, "content_tag", tag_path, warnings) + if not isinstance(product_name, str) or not isinstance(tag_name, str): + continue + key = (product_name, tag_name) + if key in lookup: + warnings.append( + f"召回明细出现重复的产品标签组合：{product_name} / {tag_name}" + ) + continue + lookup[key] = { + "count": field(tag, "count", tag_path, warnings), + "notes": as_list(tag.get("notes"), f"{tag_path}.notes", warnings), + "path": tag_path, + } + return lookup, True + + +# --------------------------------------------------------------------------- +# 召回内容提取 +# --------------------------------------------------------------------------- + +def extract_recall_notes( + detail: dict[str, Any], warnings: list[str] +) -> list[dict[str, Any]]: + notes = detail["notes"] + detail_path = detail["path"] + extracted: list[dict[str, Any]] = [] + for note_index, raw_note in enumerate(notes): + note_path = f"{detail_path}.notes[{note_index}]" + note = as_dict(raw_note, note_path, warnings) + title = note.get("title") + body = clean_full_body(note.get("desc"), title) + recall_type = note.get("type") + recall_content = field(note, "wechat_share_desc", note_path, warnings) + publish_timestamp = field(note, "time", note_path, warnings) + publish_time = format_publish_time(publish_timestamp, note_path, warnings) + if recall_type == "comment": + extracted.append({ + "召回类型": "comment", + "评论内容": recall_content, + "所属笔记点赞数量": field(note, "liked_count", note_path, warnings), + "发布时间": publish_time, + }) + continue + recall_metrics: dict[str, Any] = {} + if recall_type == "image": + images_list = as_list( + field(note, "images_list", note_path, warnings), + f"{note_path}.images_list", + warnings, + ) + recall_metrics = { + "图片数量": len(images_list), + "点赞数量": field(note, "liked_count", note_path, warnings), + "收藏数量": field(note, "collected_count", note_path, warnings), + } + elif recall_type == "video": + recall_metrics = { + "视频时长": format_video_duration( + field(note, "duration", note_path, warnings), note_path, warnings + ), + "点赞": field(note, "liked_count", note_path, warnings), + "收藏": field(note, "collected_count", note_path, warnings), + } + extracted.append({ + "序号": note_index + 1, + "召回类型": recall_type, + "笔记标题": title, + "发布时间": publish_time, + "召回内容": recall_content, + "正文内容": body, + "字数": body_char_count(body), + "评论数量": note.get("comments_count"), + **recall_metrics, + **( + {} if recall_type in {"image", "video"} else {"评论内容": None} + ), + "笔记ID": note.get("note_id"), + "笔记链接": note.get("note_url"), + }) + return extracted + + +# --------------------------------------------------------------------------- +# 核心：定位 data / data.data.complex_detail +# --------------------------------------------------------------------------- + +def locate_core_data( + root: dict[str, Any], warnings: list[str] +) -> tuple[dict[str, Any], str]: + outer_data = as_dict(root.get("data"), "data", warnings) + if isinstance(outer_data.get("complex_detail"), dict): + return outer_data, "data" + inner_data = outer_data.get("data") + if isinstance(inner_data, dict) and isinstance( + inner_data.get("complex_detail"), dict + ): + return inner_data, "data.data" + warnings.append("未在 data 或 data.data 下找到 complex_detail") + return {}, "data" + + +# --------------------------------------------------------------------------- +# 顶层 extract：对外入口 +# --------------------------------------------------------------------------- + +def extract(payload: dict[str, Any]) -> dict[str, Any]: + """ + 对爬虫原始响应 JSON 执行完整清洗，追加 result.extracted 字段。 + + 输入 payload：crawler_client 返回的原始爬虫 JSON（即存入 result_json 的内容） + 返回值：在原 payload 上追加 { "extracted": { ...清洗后结构... } } + 仅在 data.complex_detail 存在时注入 extracted + """ + warnings: list[str] = [] + root = as_dict(payload, "$", warnings) + data, base_path = locate_core_data(root, warnings) + + complex_detail = as_dict( + data.get("complex_detail"), f"{base_path}.complex_detail", warnings + ) + brand_info = as_dict( + complex_detail.get("brand_info"), + f"{base_path}.complex_detail.brand_info", + warnings, + ) + brand_sub_title = field( + brand_info, "sub_title", + f"{base_path}.complex_detail.brand_info", warnings, + ) + + # ---------- 榜单 ---------- + components = as_list( + complex_detail.get("component_list"), + f"{base_path}.complex_detail.component_list", warnings, + ) + ranking: list[dict[str, Any]] = [] + for index, raw_item in enumerate(components): + item_path = f"{base_path}.complex_detail.component_list[{index}]" + item = as_dict(raw_item, item_path, warnings) + ranking.append({ + "产品名称": field(item, "content", item_path, warnings), + "推荐比例": field(item, "desc", item_path, warnings), + "当前排名": field(item, "index", item_path, warnings), + }) + + # ---------- 产品详情 ---------- + data_list = as_list( + complex_detail.get("data_list"), + f"{base_path}.complex_detail.data_list", warnings, + ) + recall_lookup, has_recall_details = build_recall_lookup(data, base_path, warnings) + matched_recall_keys: set[tuple[str, str]] = set() + count_mismatch_total = 0 + recall_total = 0 + products: list[dict[str, Any]] = [] + for index, raw_item in enumerate(data_list): + item_path = f"{base_path}.complex_detail.data_list[{index}]" + item = as_dict(raw_item, item_path, warnings) + product_name = field(item, "text", item_path, warnings) + tag_list = as_list( + item.get("onepage_tag_list"), + f"{item_path}.onepage_tag_list", warnings, + ) + product_tags: list[dict[str, Any]] = [] + for tag_index, raw_tag in enumerate(tag_list): + tag_path = f"{item_path}.onepage_tag_list[{tag_index}]" + tag = as_dict(raw_tag, tag_path, warnings) + tag_name = field(tag, "content", tag_path, warnings) + experience_count = field(tag, "count", tag_path, warnings) + tag_output: dict[str, Any] = { + "内容标签": tag_name, + "经验数量": experience_count, + } + if ( + has_recall_details + and isinstance(product_name, str) + and isinstance(tag_name, str) + ): + key = (product_name, tag_name) + detail = recall_lookup.get(key) + if detail is None: + warnings.append(f"未找到召回明细：{product_name} / {tag_name}") + tag_output["实际召回数量"] = 0 + tag_output["召回内容"] = [] + else: + matched_recall_keys.add(key) + recall_notes = extract_recall_notes(detail, warnings) + actual_count = len(recall_notes) + recall_total += actual_count + if experience_count != actual_count: + count_mismatch_total += 1 + tag_output["实际召回数量"] = actual_count + tag_output["召回内容"] = recall_notes + product_tags.append(tag_output) + + ranked_tags = sorted( + enumerate(product_tags), + key=lambda pair: ( + numeric_sort_value(pair[1].get("经验数量")) is None, + -(numeric_sort_value(pair[1].get("经验数量")) or 0), + pair[0], + ), + ) + tag_top3 = [ + { + "排名": rank, + "内容标签": tag["内容标签"], + "经验数量": tag["经验数量"], + } + for rank, (_, tag) in enumerate(ranked_tags[:3], start=1) + ] + products.append({ + "当前排行": index + 1, + "参考经验人数": field(item, "desc", item_path, warnings), + "问点点输出": field(item, "recommend_words", item_path, warnings), + "推荐比例": field(item, "sub_title", item_path, warnings), + "产品名称": product_name, + "标签数量Top3": tag_top3, + "内容标签": product_tags, + }) + + if has_recall_details: + unmatched_keys = set(recall_lookup) - matched_recall_keys + for product_name, tag_name in sorted(unmatched_keys): + warnings.append( + f"召回明细未匹配到榜单标签：{product_name} / {tag_name}" + ) + + # ---------- 召回统计 ---------- + all_recall_notes = [ + note + for product in products + for tag in product["内容标签"] + for note in tag.get("召回内容", []) + ] + recall_type_counts: dict[str, int] = {} + for note in all_recall_notes: + recall_type = note.get("召回类型") + recall_type_counts[recall_type] = ( + recall_type_counts.get(recall_type, 0) + 1 + ) + filter_tags = as_list( + data.get("filter_tags"), f"{base_path}.filter_tags", warnings + ) + normal_names: list[Any] = [] + for index, raw_tag in enumerate(filter_tags): + tag_path = f"{base_path}.filter_tags[{index}]" + tag = as_dict(raw_tag, tag_path, warnings) + if tag.get("word_type") == "normal": + normal_names.append(field(tag, "name", tag_path, warnings)) + + extracted = { + "schema_version": "1.17", + "提取元数据": { + "包含召回明细": has_recall_details, + "召回统计": { + "标签总数": sum(len(product["内容标签"]) for product in products), + "实际召回内容总数": recall_total, + "数量不一致标签数": count_mismatch_total, + "召回类型统计": recall_type_counts, + }, + }, + "参考来源笔记总量": { + "原始文本": brand_sub_title, + "提取数量": first_integer(brand_sub_title), + }, + "榜单": ranking, + "产品详情": products, + "普通筛选标签": normal_names, + "提取警告": warnings, + } + return extracted + + +def transform_result(result: dict[str, Any]) -> dict[str, Any]: + """ + 对任务结果调用 extract 并返回合并后的结果。 + 不修改原始 result，追加 result.extracted（当爬虫返回了 complex_detail 时）。 + + 用法： + saved_result = json_loads(row["result_json"]) or {} + enriched = transform_result(saved_result) + data["result"] = enriched + """ + if not isinstance(result, dict): + return result + if "complex_detail" not in str(result.get("data", {})): + # 尝试 data.data.complex_detail（两级嵌套场景） + inner = result.get("data", {}) + if isinstance(inner, dict) and "complex_detail" not in inner: + inner = inner.get("data", {}) + if not isinstance(inner, dict) or "complex_detail" not in inner: + return result + extracted = extract(result) + return {**result, "extracted": extracted} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ac5769e --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +Flask==3.1.1 +requests==2.32.4 +python-dotenv==1.1.1 +gunicorn==23.0.0 diff --git a/rule_service.py b/rule_service.py new file mode 100644 index 0000000..526ac35 --- /dev/null +++ b/rule_service.py @@ -0,0 +1,103 @@ +from pathlib import Path +from typing import Any + +from flask import Request + +from config import Config +from database import json_dumps, json_loads, now_iso +from task_service import TaskError + + +RULE_DIR = Config.FILE_STORAGE_PATH / "rules" +RULE_HTML_PATH = RULE_DIR / "current.html" +RULE_META_PATH = RULE_DIR / "current.json" + + +def _ensure_rule_dir() -> None: + RULE_DIR.mkdir(parents=True, exist_ok=True) + + +def _read_text(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def get_rule_document() -> dict[str, Any]: + """读取当前规则 HTML 的公开元数据。""" + + if not RULE_HTML_PATH.exists(): + return { + "exists": False, + "title": "", + "filename": "", + "updated_at": None, + "size_bytes": 0, + "html_url": "/api/v1/rules/current.html", + } + metadata = json_loads(_read_text(RULE_META_PATH)) if RULE_META_PATH.exists() else {} + stat = RULE_HTML_PATH.stat() + return { + "exists": True, + "title": metadata.get("title") or "小红书问一问收录规则探索", + "filename": metadata.get("filename") or RULE_HTML_PATH.name, + "updated_at": metadata.get("updated_at"), + "size_bytes": stat.st_size, + "html_url": "/api/v1/rules/current.html", + } + + +def get_rule_html_path() -> Path: + if not RULE_HTML_PATH.exists(): + raise TaskError("RULE_NOT_FOUND", "当前没有已上传的规则 HTML", 404) + return RULE_HTML_PATH + + +def save_rule_document(request: Request) -> dict[str, Any]: + """保存当前规则 HTML；支持 multipart 文件、JSON html 字段和 text/html 原文。""" + + filename = "current.html" + title = "小红书问一问收录规则探索" + html = "" + uploaded = request.files.get("file") + if uploaded: + filename = uploaded.filename or filename + html = uploaded.read().decode("utf-8") + title = request.form.get("title") or title + else: + payload = request.get_json(silent=True) + if isinstance(payload, dict): + html = str(payload.get("html") or "") + filename = str(payload.get("filename") or filename) + title = str(payload.get("title") or title) + else: + html = request.get_data(as_text=True) or "" + title = request.args.get("title") or title + filename = request.args.get("filename") or filename + + if not html.strip(): + raise TaskError("RULE_HTML_REQUIRED", "html 内容不能为空", 400) + if " dict[str, Any]: + """删除当前规则 HTML 和元数据。""" + + deleted = False + for path in (RULE_HTML_PATH, RULE_META_PATH): + if path.exists(): + path.unlink() + deleted = True + return {"deleted": deleted, **get_rule_document()} diff --git a/scripts/register_token.py b/scripts/register_token.py new file mode 100644 index 0000000..baf5c5d --- /dev/null +++ b/scripts/register_token.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python3 +""" +register_token.py — 将用户的 hc_session token 注册到 AI Token 表， + 供 AI 直接调用后端 API。 + +用户只需提供一次，后端存储后 AI 后续就能用这个 token 访问。 + +用法： + python3 register_token.py + python3 register_token.py # 静默模式（用于自动化） +""" + +import hashlib +import json +import os +import secrets +import sys +import webbrowser +from datetime import datetime, timedelta +from pathlib import Path +from urllib.parse import urlencode + +# ---------- 项目路径 ---------- +SCRIPT_DIR = Path(__file__).resolve().parent +BACKEND_DIR = SCRIPT_DIR.parent +sys.path.insert(0, str(BACKEND_DIR)) + +from config import Config +from database import connection, now_iso + + +def _sha256(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _normalize_user_from_token(token: str) -> dict | None: + """用 token 查找用户，不存在则返回 None""" + session_hash = _sha256(token) + current = now_iso() + with connection() as conn: + row = conn.execute( + """SELECT s.id AS session_id,u.* FROM auth_sessions s + JOIN auth_users u ON u.id=s.user_id + WHERE s.session_hash=? AND s.revoked_at IS NULL AND s.expires_at>?""", + (session_hash, current), + ).fetchone() + if not row: + return None + conn.execute( + "UPDATE auth_sessions SET last_seen_at=? WHERE id=?", + (current, row["session_id"]), + ) + return { + "id": str(row["ding_user_id"]), + "name": row["name"], + "ding_user_id": row["ding_user_id"], + } + + +def ensure_ai_tokens_table() -> None: + """建 AI Token 表（如果不存在）""" + with connection() as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS ai_access_tokens ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + token_hash TEXT NOT NULL UNIQUE, + user_id TEXT NOT NULL, + user_name TEXT NOT NULL, + description TEXT DEFAULT '', + created_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + revoked_at TEXT, + last_used_at TEXT, + use_count INTEGER NOT NULL DEFAULT 0 + ) + """) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_ai_token_hash ON ai_access_tokens(token_hash)" + ) + + +def register_ai_token(token: str, user_info: dict, description: str = "") -> dict: + """注册 / 更新 AI 访问 token""" + token_hash = _sha256(token) + now = now_iso() + expires = (datetime.now().astimezone() + timedelta(days=365)).isoformat( + timespec="seconds" + ) + with connection() as conn: + existing = conn.execute( + "SELECT id FROM ai_access_tokens WHERE token_hash=?", + (token_hash,), + ).fetchone() + if existing: + conn.execute( + """UPDATE ai_access_tokens + SET last_used_at=?, use_count=use_count+1 + WHERE token_hash=?""", + (now, token_hash), + ) + action = "updated" + else: + conn.execute( + """INSERT INTO ai_access_tokens + (token_hash,user_id,user_name,description,created_at,expires_at,last_used_at,use_count) + VALUES (?,?,?,?,?,?,?,1)""", + ( + token_hash, user_info["ding_user_id"], + user_info["name"], description, now, expires, now, + ), + ) + action = "registered" + return {"action": action, "user": user_info["name"], "expires_at": expires} + + +def revoke_ai_token(token: str) -> bool: + token_hash = _sha256(token) + with connection() as conn: + conn.execute( + "UPDATE ai_access_tokens SET revoked_at=? WHERE token_hash=? AND revoked_at IS NULL", + (now_iso(), token_hash), + ) + affected = conn.total_changes + return affected > 0 + + +def list_ai_tokens() -> list[dict]: + with connection() as conn: + rows = conn.execute( + """SELECT user_name,description,created_at,expires_at,last_used_at,use_count,revoked_at + FROM ai_access_tokens WHERE revoked_at IS NULL + ORDER BY last_used_at DESC""" + ).fetchall() + return [dict(r) for r in rows] + + +# ---------- 主入口：交互式注册流程 ---------- +def main() -> None: + ensure_ai_tokens_table() + + backend = os.environ.get("HC_BACKEND_URL", f"http://localhost:{Config.PORT}").rstrip("/") + + print() + print("═" * 55) + print(" 🔐 Holy Crab · AI Token 注册") + print("═" * 55) + print() + print(" 本工具用于将你的 hc_session token 注册给 AI 访问权限。") + print() + print(" Step 1：请从浏览器获取你的 hc_session token") + print() + print(" 操作步骤：") + print(" 1. 在浏览器中打开 Holy Crab 前端并完成钉钉登录") + print(" 2. 打开浏览器 DevTools (F12) → Application → Cookies") + print(" 3. 找到你的后端地址，点击展开 Cookie 列表") + print(" 4. 复制 'hc_session' 的 Value（很长的一串字符）") + print() + + # 检查是否传入参数 + if len(sys.argv) > 1: + token = sys.argv[1].strip() + if token in ("--help", "-h"): + print("用法: python3 register_token.py [token]") + sys.exit(0) + print(f" 检测到命令行传入 token: {token[:8]}...{token[-4:]}") + else: + print(" 直接回车打开浏览器说明网页（推荐）：") + print() + print(" 或者手动输入 token（粘贴后回车）：") + token = input(" > ").strip() + + if not token: + # 打开说明网页 + help_url = ( + f"{backend}/api/v1/auth/dingtalk/login" + f"?redirect={urlencode({'r': '/__help__'})}" + ) + print() + print(f" 正在打开说明网页：{help_url}") + webbrowser.open(help_url) + print() + print(" 请按上述步骤获取 token 后再次运行：") + print(f" python3 {sys.argv[0]} ") + sys.exit(0) + + print() + print(" 🔍 正在验证 token...") + user = _normalize_user_from_token(token) + if not user: + print() + print(" ❌ token 无效或已过期") + print(" 请确认：") + print(" 1. token 是否正确（注意不要有前后空格）") + print(" 2. 是否已通过钉钉登录") + print(" 3. token 是否已过期（默认14天）") + sys.exit(1) + + print(f" ✅ 验证成功！用户：{user['name']}") + print() + result = register_ai_token( + token, user, + description=f"AI助手访问 (注册IP/来源: register_script)" + ) + print(f" ✅ AI Token 注册成功") + print(f" 用户：{result['user']}") + print(f" 有效期至：{result['expires_at']}") + print() + print(" 📋 AI 后续使用方式：") + print() + print(" 方式 A — 设置环境变量（推荐，每次会话有效）：") + print(f" export HC_SESSION='{token}'") + print() + print(" 方式 B — 写入文件（持久化）：") + print(f" echo '{token}' > ~/.holy_crab_cookie") + print() + print(" 方式 C — 写入配置（自动读取）：") + print(f" echo 'HC_SESSION={token}' >> ~/.holy_crab_env") + print(" # scripts 会自动从 ~/.holy_crab_env 读取环境变量") + print() + print("═" * 55) + print() + + +if __name__ == "__main__": + main() diff --git a/skills/holy-crab/PROMPTS.md b/skills/holy-crab/PROMPTS.md new file mode 100644 index 0000000..500db04 --- /dev/null +++ b/skills/holy-crab/PROMPTS.md @@ -0,0 +1,92 @@ +# Holy Crab · AI 可直接使用的提示词 + +## 用法一：作为 System Prompt（给任意 AI 使用） + +将以下内容粘贴到 AI 的 System Prompt 或 Custom Instructions 中： + +--- + +``` +## Holy Crab 小红书问一问数据助手 + +你是 Holy Crab 搜索监测平台的数据分析助手。用户授权后，你可以直接访问后端 API 查询任务数据、生成分析报告。 + +### 认证方式（首次使用） +用户提供钉钉授权码（authCode），你调用接口换取 session token 并保存： + +1. 用户说「授权」→ 你执行： + bash ~/.qclaw/skills/holy-crab/scripts/get-cookie.sh +2. 用户把浏览器 URL 里的 authCode=xxxxx 复制给你 +3. 你执行： + bash ~/.qclaw/skills/holy-crab/scripts/get-cookie.sh +4. Token 自动保存，后续无需再授权 + +后端地址（按需设置）： + export HC_BACKEND_URL=http://localhost:8000 + +### 数据查询命令 +# 查看任务列表 +python3 ~/.qclaw/skills/holy-crab/scripts/query_tasks.py + +# 获取任务详情（含 extracted 结构化数据） +python3 ~/.qclaw/skills/holy-crab/scripts/query_tasks.py <任务ID> + +# 数据分析 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py summarize <任务ID> # 榜单摘要 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py sentiment <任务ID> # 评论情感 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py compare <任务ID> # 产品对比 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py detail <任务ID> # 完整JSON + +### extracted 数据解读 +extracted 是清洗后的结构化数据，关键字段： +- schema_version：数据版本（当前 1.17） +- 参考来源笔记总量：原始笔记数量 +- 榜单：产品排名列表，含产品名称、推荐比例 +- 产品详情：含标签数量Top3、内容标签（含召回的笔记/评论） +- 普通筛选标签：关键词标签 +- 提取元数据：召回统计（标签总数、实际召回总数） +- 提取警告：解析警告列表 + +### 报告生成 +根据 extracted 数据生成 markdown 格式分析报告，包含： +- 任务概况（关键词、模式、数据规模） +- 榜单排名分析（表格 + 解读） +- 产品详细分析（标签、召回内容摘要） +- 评论情感分析（语义判断，补充情感分类） +- 产品横向对比（关键指标对比表） +- 提炼与建议（主要发现、市场机会、内容建议） + +### 情感判断规则 +评论（召回类型=comment）的情感分类需 AI 语义判断： +- 正面：赞美、推荐、效果满意、回购意向 +- 负面：吐槽、投诉、效果差、不推荐 +- 中性：疑问、客观描述、无明确情感倾向 + +### 错误处理 +- Token 无效/过期 → 重新引导用户授权 +- 任务不存在 → 检查 task_id +- extracted 为空 → 任务未完成，告知用户 +- 评论数据为空 → 轻度/常规任务无评论，建议深度模式 +``` + +--- + +## 用法二：给用户的一句话指令 + +直接复制给 AI 使用： + +> **「请帮我分析 Holy Crab 的 [任务ID] 任务，生成一份完整的分析报告，包含榜单、产品对比和情感分析。」** +> +> （AI 会自动引导你完成首次授权，之后直接出报告） + +--- + +## 用法三：用户操作简化版 + +``` +# 第一次使用，告诉 AI： +帮我分析任务结果，授权码是：<你的authCode> + +# 之后每次直接说： +生成报告，任务ID是 HC-20260727-XXXXXX +``` diff --git a/skills/holy-crab/SKILL.md b/skills/holy-crab/SKILL.md new file mode 100644 index 0000000..d772603 --- /dev/null +++ b/skills/holy-crab/SKILL.md @@ -0,0 +1,242 @@ +--- +name: holy-crab +description: "Holy蟹搜索监测平台 — 当用户需要查询/分析小红书问一问任务结果、榜单数据、产品详情、评论情感时使用此技能。包含结果查询、榜单摘要、产品分析、情感统计等完整处理流程。" +--- + +# Holy蟹搜索监测 · AI 数据访问技能 + +## 概述 + +Holy Crab（Holy蟹）是一个基于小红书问一问 API 的搜索监测平台。 +爬虫结果经 `extract_wenyiwen.py` 清洗后，以标准结构化格式存储。 + +**技能触发词**： +- URL 中含 `task_id=`（用户从 Holy Crab 前端点击「Link AI」跳转而来，参数从 URL 或首条消息中解析） +- 查一下任务、看任务结果、榜单分析、产品分析 +- 问一问数据、情感分析、提取评论 +- Holy蟹、holy crab、Holy Crab + +**自动触发流程**（从前端 Link AI 跳转时）： +1. 从 URL 参数或首条用户消息中提取 `task_id` +2. 自动调用 `python3 scripts/query_tasks.py ` 获取任务数据 +3. 若有 `task_name`/`keywords`/`mode` 参数，合并到报告上下文中 +4. 生成完整分析报告 + +--- + +## 快速开始 + +### 第 1 步：获取会话 Cookie（authCode 方式） + +Holy Crab 后端使用钉钉 OAuth，AI 只需用户的 authCode 就能换出 session token，**全程不需要浏览器 Cookie**。 + +#### 自动流程（推荐）： +```bash +bash scripts/get-cookie.sh +# 会自动打开钉钉授权页面，引导用户完成授权 +# 授权后浏览器 URL 带 authCode，用户把 authCode 贴给 AI +# AI 调用 /api/v1/auth/direct-token 换出 session token，存入 ~/.holy_crab_env +``` + +#### AI 持有 token 后的用法 + +用户告诉 AI authCode： +``` +【authCode】YOUR_AUTH_CODE +``` + +AI 自动执行： +```bash +export HC_SESSION=$(curl -s "http://localhost:8000/api/v1/auth/direct-token?authCode=YOUR_AUTH_CODE" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['token'])") +``` + +后续所有请求自动带 Cookie，无需用户再次操作。 + +```bash +# 先确保后端正在运行，然后执行： +bash ~/.qclaw/skills/holy-crab/scripts/get-cookie.sh +# 成功后会输出：✅ Cookie 已写入 ~/.holy_crab_cookie +``` + +> **如果自动方式失败**（后端未启动或网络不通），改为手动方式： +> 1. 在浏览器中登录 Holy Crab 前端 +> 2. 打开浏览器 DevTools → Application → Cookies，复制 `hc_session` 的值 +> 3. 写入文件：`echo "你的cookie值" > ~/.holy_crab_cookie` + +### 第 2 步：查询任务列表 + +```bash +python3 ~/.qclaw/skills/holy-crab/scripts/query_tasks.py +# 输出：任务ID | 名称 | 关键词 | 状态 | 进度 | 创建时间 +``` + +### 第 3 步：获取任务详情（含 extracted 数据） + +```bash +python3 ~/.qclaw/skills/holy-crab/scripts/query_task.py <任务ID> +``` + +返回示例（`result.extracted` 部分）： +```json +{ + "schema_version": "1.17", + "参考来源笔记总量": { "原始文本": "5000条内容", "提取数量": 5000 }, + "榜单": [ + { "当前排名": 1, "产品名称": "某品牌A", "推荐比例": "42%" } + ], + "产品详情": [ + { + "当前排行": 1, + "产品名称": "某品牌A", + "参考经验人数": "300人体验", + "推荐比例": "42%", + "标签数量Top3": [ + { "排名": 1, "内容标签": "保湿效果好", "经验数量": 150 } + ], + "内容标签": [ + { + "内容标签": "保湿效果好", + "经验数量": 150, + "实际召回数量": 8, + "召回内容": [ + { + "召回类型": "image", + "笔记标题": "实测分享", + "正文内容": "用了两周皮肤确实变好了...", + "字数": 120, + "点赞数量": 234, + "收藏数量": 56, + "笔记链接": "http://xhslink.com/..." + }, + { + "召回类型": "comment", + "评论内容": "真的好用！", + "所属笔记点赞数量": 234, + "发布时间": "2026-07-20 14:30:00" + } + ] + } + ] + } + ], + "普通筛选标签": ["成分党", "敏感肌"], + "提取警告": [] +} +``` + +### 第 4 步：调用处理函数 + +```bash +# 榜单摘要 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py summarize <任务ID> + +# 情感统计（从召回内容中提取 comment 类型） +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py sentiment <任务ID> + +# 产品对比 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py compare <任务ID> +``` + +--- + +## API 参考 + +### 基础信息 + +| 项目 | 值 | +|------|-----| +| 后端地址 | `http://localhost:8000`（本地）或 `http://<服务器IP>:8000` | +| API 前缀 | `/api/v1` | +| 认证方式 | Session Cookie（`hc_session`） | + +### 核心接口 + +#### GET /api/v1/tasks +任务列表（支持 `?q=关键词&status=completed&page=1&page_size=50`） + +#### GET /api/v1/tasks/{task_id} +单个任务详情（含 `result.extracted`） + +#### GET /api/v1/tasks/{task_id}/files +任务文件列表 + +--- + +## result.extracted 数据解读 + +### 顶层字段 + +| 字段 | 类型 | 说明 | +|------|------|------| +| `schema_version` | string | 数据版本，当前 `1.17` | +| `参考来源笔记总量` | object | brand_info 中解析的笔记总数 | +| `榜单` | array | 推荐产品排名列表 | +| `产品详情` | array | 各产品的详细标签和召回内容 | +| `普通筛选标签` | array | 普通命中关键词列表 | +| `提取元数据` | object | 包含 `召回统计` 和 `包含召回明细` | +| `提取警告` | array | 解析过程中的警告信息 | + +### 榜单元素 + +```json +{ "当前排名": 1, "产品名称": "某品牌", "推荐比例": "42%" } +``` + +### 产品详情元素 + +| 字段 | 说明 | +|------|------| +| `当前排行` | 在榜单中的排名 | +| `产品名称` | 产品名称 | +| `参考经验人数` | 原始文本，如 "300人体验" | +| `推荐比例` | 在问一问中的推荐占比 | +| `标签数量Top3` | 按经验数量排序的前3个标签 | +| `内容标签` | 全部内容标签，含召回内容 | + +### 召回内容类型 + +| `召回类型` | 含义 | 额外字段 | +|-----------|------|---------| +| `image` | 图文笔记 | `笔记标题`、`正文内容`、`字数`、`点赞数量`、`收藏数量`、`评论数量`、`笔记链接` | +| `video` | 视频笔记 | 同 image + `视频时长` | +| `comment` | 评论召回 | `评论内容`、`所属笔记点赞数量`、`发布时间` | + +### 情感统计（需 AI 自行计算） + +从 `产品详情[].内容标签[].召回内容[]` 中筛选 `召回类型 == "comment"` 的条目， +统计 `情感分类` 字段（正面/负面/中性）。 + +> **注意**：`情感分类` 字段需要通过 AI 语义分析补充，原始清洗数据中不含此字段。 +> AI 应读取 `评论内容`，判断情感后补充统计。 + +--- + +## 认证机制详解 + +Holy Crab 使用钉钉 OAuth + Cookie 会话： + +1. 用户访问 `/api/v1/auth/dingtalk/login` → 跳转到钉钉授权 +2. 钉钉回调 `/api/v1/auth/dingtalk/callback` → 设置 `hc_session` Cookie +3. 后续请求携带该 Cookie 访问所有 `/api/v1/*` 接口 + +**AI 获取 Cookie 的方法**（按优先级）： + +1. **环境变量**：`HC_SESSION=xxx`（最优先） +2. **Cookie 文件**：`~/.holy_crab_cookie` +3. **直接提示用户**提供 Cookie + +```bash +# 设置环境变量方式 +export HC_SESSION="用户提供的cookie值" +``` + +--- + +## 错误处理 + +| 错误信息 | 含义 | 处理方式 | +|---------|------|---------| +| `请先通过钉钉授权登录` | Cookie 无效或过期 | 提示用户重新登录后端或刷新 Cookie | +| `任务不存在` | task_id 有误 | 检查任务 ID 是否正确 | +| `服务器内部错误` | 后端异常 | 检查后端服务是否运行 | +| `提取警告` 数组有内容 | 数据解析有部分问题 | 查看 warning 字段了解详情 | diff --git a/skills/holy-crab/SYSTEM_PROMPT.md b/skills/holy-crab/SYSTEM_PROMPT.md new file mode 100644 index 0000000..11ffd68 --- /dev/null +++ b/skills/holy-crab/SYSTEM_PROMPT.md @@ -0,0 +1,191 @@ +# Holy Crab · AI 数据分析报告生成提示词 + +> 本提示词供 AI 助手使用，当用户需要查询、分析或生成 Holy Crab 任务报告时自动激活。 +> AI 应严格按照以下步骤执行，除非用户明确指定了不同流程。 + +--- + +## 角色与职责 + +你是一个专业的 **小红书问一问数据分析助手**，擅长： +- 读取 Holy Crab 后端结构化数据 +- 解读 `result.extracted` 数据字段 +- 生成专业的市场/竞品分析报告 +- 支持榜单分析、产品对比、情感统计等多种报告类型 + +--- + +## 认证流程（每次会话首次使用时执行） + +### Step 0：检测是否从前端 Link AI 跳转而来（优先执行） + +当用户发送的第一条消息包含 URL 参数，或消息本身以 `task_id=` 开头时： + +1. 从消息中提取 `task_id`、`task_name`、`keywords`、`mode` 参数 + - 如果是 URL：`https://content.gbotai.cn/?task_id=xxx&keywords=yyy` → 解析参数 + - 如果是纯文本消息：查找 `task_id=xxx` 等 key=value 对 +2. 若提取到 `task_id`，直接执行 Step 3-5，无需额外确认 +3. 生成报告时在开头注明「数据来源：用户从 Holy Crab 前端跳转，任务 ID 由 URL 参数提供」 + +### Step 1：检查是否有有效 Token + +```bash +bash ~/.qclaw/skills/holy-crab/scripts/get-cookie.sh --check +``` + +- **返回 ✅ Token 有效** → 跳过认证，直接进入 Step 3 +- **返回 ❌ 未找到有效 Token** → 进入 Step 2 + +### Step 2：引导用户完成钉钉授权 + +```bash +bash ~/.qclaw/skills/holy-crab/scripts/get-cookie.sh +``` + +按脚本输出引导用户： +1. 脚本会自动打开钉钉授权页（或显示 URL 让用户手动打开） +2. 用户在浏览器完成钉钉授权 +3. 授权后浏览器 URL 带 `authCode=xxxxx`，用户复制这串字符 +4. 用户告诉 AI：「authCode 是 xxxxx」 +5. AI 执行：`bash ~/.qclaw/skills/holy-crab/scripts/get-cookie.sh ` +6. Token 自动存入 `~/.holy_crab_env`，后续请求自动带 Cookie + +> **注意**：authCode 有效期约 60 秒，必须在生成后立即使用。若过期，让用户重新授权。 + +--- + +## 数据查询流程 + +### Step 3：查看任务列表 + +```bash +python3 ~/.qclaw/skills/holy-crab/scripts/query_tasks.py +``` + +输出格式： +``` +任务ID | 状态 | 关键词 | 创建时间 | 文件数 +``` +找到用户要分析的任务，记录其 `task_id`。 + +### Step 4：获取任务详情 + +```bash +python3 ~/.qclaw/skills/holy-crab/scripts/query_tasks.py <任务ID> +``` + +输出包含： +- 基本信息（状态、进度、创建/完成时间） +- `extracted` 数据清洗结果（schema_version、榜单、产品数量、召回统计） + +### Step 5：数据分析 + +| 分析目的 | 执行命令 | +|---------|---------| +| 榜单摘要 | `python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py summarize <任务ID>` | +| 评论情感统计 | `python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py sentiment <任务ID>` | +| 产品横向对比 | `python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py compare <任务ID>` | +| 完整 JSON | `python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py detail <任务ID>` | + +> **情感分析补充**：原始数据中评论无 `情感分类` 字段，AI 应读取 `评论内容`，通过语义判断补充情感标签（正面/负面/中性），再统计各情感占比。 + +--- + +## 报告生成规范 + +### 报告结构模板 + +``` +# 【项目名称】小红书问一问监测报告 +> 生成时间 | 任务ID | 数据来源说明 + +## 一、任务概况 +- 监测关键词 +- 采集模式（轻度/常规/深度） +- 数据规模（参考笔记总量、产品数量、标签数量） +- 采集时间范围 + +## 二、榜单排名分析 +- 榜单概览表（排名、产品名称、推荐比例、经验人数） +- 各产品市场占比可视化描述 +- Top 标签分布解读 + +## 三、产品详细分析 +对每个上榜产品分别描述： +- 产品基本信息 +- 用户标签 Top3（按经验数量排序） +- 推荐理由归纳 +- 召回内容摘要（笔记类型分布、文字/视频占比） + +## 四、评论情感分析 +> 仅深度任务有评论数据，轻度/常规任务可跳过此节 +- 评论总量 +- 情感分布（正面/负面/中性 各占比） +- 代表性评论摘录 +- 用户反馈洞察 + +## 五、产品横向对比 +- 关键指标对比表 +- 各产品优劣势总结 +- 差异化标签分析 + +## 六、提炼与建议 +- 主要发现（3~5 条） +- 市场机会点 +- 内容营销建议 +- 后续监测建议 + +## 附录：数据质量说明 +- 原始笔记总量 vs 实际召回数量 +- 提取警告（如有） +- schema_version +``` + +### 报告输出要求 + +1. **语言**：中文，markdown 格式，支持表格 +2. **数据来源**：所有数字必须来自 `extracted` 字段，不得虚构 +3. **情感判断**：评论情感需 AI 语义分析后补充，说明判断依据 +4. **数据缺失**：若某项数据为空（如轻度任务无评论），明确标注「数据不足/不适用」 +5. **字数**：完整报告 800~2000 字，摘要版 400~600 字 + +--- + +## 快速分析模式（用户只问简单问题） + +当用户的问题比较简单（如「哪个产品排名最高」「有多少条评论」），直接： + +```bash +python3 ~/.qclaw/skills/holy-crab/scripts/query_tasks.py <任务ID> # 查看任务摘要 +python3 ~/.qclaw/skills/holy-crab/scripts/process_data.py summarize <任务ID> # 榜单摘要 +``` + +根据输出结果直接回答，不需要生成完整报告。 + +--- + +## 错误处理 + +| 场景 | 处理方式 | +|------|---------| +| 「请先通过钉钉授权登录」| Token 过期，重新执行 Step 2 | +| 「任务不存在」| 检查 task_id 是否正确 | +| 「后端不可达」| 确认后端是否启动，HC_BACKEND_URL 是否正确 | +| extracted 数据为空 | 任务可能未完成，告知用户等待或检查任务状态 | +| 评论数据为空 | 轻度/常规任务默认无评论召回，说明原因，建议用深度模式重新采集 | + +--- + +## 环境变量参考 + +| 变量名 | 说明 | 默认值 | +|--------|------|--------| +| `HC_BACKEND_URL` | 后端地址 | `http://localhost:8000` | +| `HC_SESSION` | 会话 Token（环境变量优先） | — | +| `~/.holy_crab_env` | Token 持久化文件 | — | +| `~/.holy_crab_cookie` | Token 文件（兜底） | — | + +设置后端地址示例： +```bash +export HC_BACKEND_URL=http://服务器IP:8000 +``` diff --git a/skills/holy-crab/VERSION b/skills/holy-crab/VERSION new file mode 100644 index 0000000..9fe2c9b --- /dev/null +++ b/skills/holy-crab/VERSION @@ -0,0 +1,5 @@ +{ + "version": "1.1.0", + "released": "2026-07-27", + "changelog": "1.1.0: 支持前端 Link AI 跳转（URL参数自动触发）+ check_update版本管理。1.0.0: 初始版本"} +} diff --git a/skills/holy-crab/scripts/check_update.py b/skills/holy-crab/scripts/check_update.py new file mode 100755 index 0000000..5003b84 --- /dev/null +++ b/skills/holy-crab/scripts/check_update.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +""" +check_update.py — 检查 / 更新 Holy Crab Skill 版本 + +用法： + python3 check_update.py # 检查更新 + python3 check_update.py --force # 强制更新到最新版本 + python3 check_update.py --install # 从后端下载并安装 +""" + +import hashlib +import json +import os +import shutil +import sys +import zipfile +from datetime import datetime +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +SKILL_DIR = SCRIPT_DIR.parent +VERSION_FILE = SKILL_DIR / "VERSION" +BACKEND = os.environ.get("HC_BACKEND_URL", "http://localhost:8000").rstrip("/") +VERSION_API = f"{BACKEND}/api/v1/skill/holy-crab/version" +DOWNLOAD_API = f"{BACKEND}/api/v1/skill/holy-crab/download" + + +def local_version() -> dict | None: + if not VERSION_FILE.exists(): + return None + return json.loads(VERSION_FILE.read_text()) + + +def compute_hash(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest()[:16] + + +def update_version_file(remote: dict) -> None: + # 更新本地 VERSION + v = remote.copy() + for fname, finfo in v.get("files", {}).items(): + fpath = SKILL_DIR / fname + if fpath.exists(): + finfo["hash"] = compute_hash(fpath) + VERSION_FILE.write_text(json.dumps(v, ensure_ascii=False, indent=2)) + + +def check_update() -> tuple[bool, dict | None, dict | None]: + """检查是否有更新。返回 (有更新, 本地版本, 远程版本)""" + local = local_version() + remote_v = None + + import urllib.request + try: + req = urllib.request.Request(VERSION_API, headers={"Accept": "application/json"}) + with urllib.request.urlopen(req, timeout=10) as resp: + raw = json.loads(resp.read().decode("utf-8")) + # 后端返回格式：{"data": {"version": {...}}} + remote_v = raw.get("data", {}).get("version") or raw + except Exception: + return False, local, None + + if not local: + return True, None, remote_v + + def parse_ver(v: str) -> tuple: + return tuple(int(x) for x in v.split(".")) + + local_ver = local.get("version", "0.0.0") + remote_ver = remote_v.get("version", "0.0.0") + has_update = parse_ver(remote_ver) > parse_ver(local_ver) + return has_update, local, remote_v + + +def download_and_install() -> None: + import urllib.request + print(" 📥 正在从后端下载...") + try: + req = urllib.request.Request(DOWNLOAD_API, headers={"Accept": "application/zip"}) + with urllib.request.urlopen(req, timeout=60) as resp: + zip_data = resp.read() + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + print(f" ❌ 下载失败 HTTP {e.code}: {body[:200]}") + sys.exit(1) + + # 备份当前版本 + backup_dir = SKILL_DIR.parent / f"holy-crab.backup.{datetime.now().strftime('%Y%m%d%H%M%S')}" + shutil.copytree(SKILL_DIR, backup_dir) + print(f" 📦 当前版本已备份至: {backup_dir}") + + # 解压到 skill 目录 + import io + with zipfile.ZipFile(io.BytesIO(zip_data), "r") as zf: + members = [m for m in zf.namelist() if not m.endswith("/")] + for member in members: + # 去掉顶层目录名（skill 安装包通常带 holy-crab/v1.0.0/ 前缀） + dst_name = member + for prefix in [f"holy-crab-{VERSION_FILE.parent.name}/", "holy-crab/"]: + if member.startswith(prefix): + dst_name = member[len(prefix):] + break + if not dst_name or dst_name == member: + # 直接从根目录解压的文件 + dst_name = member.split("/", 1)[1] if "/" in member else member + + dst = SKILL_DIR / dst_name + dst.parent.mkdir(parents=True, exist_ok=True) + if dst_name: + dst.write_bytes(zf.read(member)) + + print(" ✅ 安装完成！") + + +def main() -> None: + force = "--force" in sys.argv + install = "--install" in sys.argv or "-u" in sys.argv + + print() + print("═" * 55) + print(" 🔍 Holy Crab Skill 版本检查") + print("═" * 55) + print() + print(f" 后端地址: {BACKEND}") + print(f" 本地路径: {SKILL_DIR}") + print() + + local = local_version() + if local: + print(f" 本地版本: {local.get('version')} ({local.get('released', '未知日期')})") + else: + print(" 本地版本: 未安装") + + has_update, _, remote = check_update() + + if not remote: + print() + print(" ⚠️ 无法连接到后端获取远程版本信息") + print(" 请确认：") + print(" 1. 后端是否已启动") + print(" 2. HC_BACKEND_URL 是否正确") + print() + sys.exit(1) + + print(f" 远程版本: {remote.get('version')} ({remote.get('released', '未知日期')})") + + if not has_update and not force: + print() + print(" ✅ 当前已是最新版本，无需更新") + print() + if local: + print(" 文件状态:") + for fname, finfo in local.get("files", {}).items(): + fpath = SKILL_DIR / fname + status = "✅" if fpath.exists() else "❌ 缺失" + print(f" {status} {fname}") + print() + return + + print() + print(f" 🆕 发现新版本: {remote.get('version')}") + changelog = remote.get("changelog", "") + if changelog: + print(f" 更新内容: {changelog}") + print() + + if install or force or "--yes" in sys.argv: + confirm = True + else: + answer = input(" 是否立即更新？(y/n): ").strip().lower() + confirm = answer in ("y", "yes", "是") + + if confirm: + download_and_install() + # 更新本地 VERSION 记录 + if local_version(): + update_version_file(remote) + print() + print(f" ✅ Holy Crab Skill 已更新到 {remote.get('version')}") + else: + print(" 已取消更新") + + print() + print("═" * 55) + + +if __name__ == "__main__": + main() diff --git a/skills/holy-crab/scripts/get-cookie.sh b/skills/holy-crab/scripts/get-cookie.sh new file mode 100755 index 0000000..38dc4c4 --- /dev/null +++ b/skills/holy-crab/scripts/get-cookie.sh @@ -0,0 +1,216 @@ +#!/usr/bin/env bash +# get-cookie.sh — Holy Crab AI 认证流程 +# +# 流程： +# 1. 从后端获取钉钉授权 URL +# 2. 让用户在浏览器打开，授权后 URL 里会带 authCode +# 3. AI 把 authCode 贴过来 → 调用 /api/v1/auth/direct-token 换出 session token +# 4. 存 token 到 ~/.holy_crab_env，后续所有请求自动带 Cookie +# +# 用法： +# bash get-cookie.sh # 交互式 +# bash get-cookie.sh --check # 仅检查 token 是否有效 +# bash get-cookie.sh # 直接用 authCode 换 token + +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +SKILL_DIR="$(dirname "$SCRIPT_DIR")" +COOKIE_FILE="$HOME/.holy_crab_cookie" +ENV_FILE="$HOME/.holy_crab_env" +BACKEND="${HC_BACKEND_URL:-http://localhost:8000}" + +# --------------------------------------------------------------------------- +# 工具函数 +# --------------------------------------------------------------------------- + +save_token() { + local token="$1" + echo "HC_SESSION=${token}" > "$ENV_FILE" + chmod 600 "$ENV_FILE" + echo "✅ Token 已写入: $ENV_FILE" + echo "" + echo "📋 后续 AI 将自动从此文件读取 Cookie，无需任何额外操作。" +} + +save_token_raw() { + local token="$1" + echo "$token" > "$COOKIE_FILE" + chmod 600 "$COOKIE_FILE" +} + +check_token() { + local token="$1" + local http_code + http_code=$(curl -s -o /dev/null -w "%{http_code}" \ + --max-time 10 \ + -H "Cookie: hc_session=${token}" \ + "$BACKEND/api/v1/auth/me") + if [ "$http_code" = "200" ]; then + echo "✅ Token 有效" + return 0 + else + echo "❌ Token 无效或已过期 (HTTP $http_code)" + return 1 + fi +} + +# 从现有文件读 token +load_existing_token() { + # 优先 ~/.holy_crab_env + if [ -f "$ENV_FILE" ]; then + local val + val=$(grep -E "^HC_SESSION=" "$ENV_FILE" 2>/dev/null | head -1 | cut -d= -f2- | tr -d '"' | tr -d "'" | xargs) + [ -n "$val" ] && echo "$val" && return 0 + fi + # 兜底 ~/.holy_crab_cookie + if [ -f "$COOKIE_FILE" ]; then + local val + val=$(cat "$COOKIE_FILE" | xargs) + [ -n "$val" ] && echo "$val" && return 0 + fi + return 1 +} + +# --------------------------------------------------------------------------- +# Step 1：检查现有 token +# --------------------------------------------------------------------------- + +if [ "$1" = "--check" ]; then + echo "" + echo "🔍 检查现有 Token..." + if EXISTING=$(load_existing_token 2>/dev/null); then + if check_token "$EXISTING"; then + echo "" + echo " Token 来自: ${ENV_FILE:-$COOKIE_FILE}" + exit 0 + fi + fi + echo " 未找到有效 Token，需要重新授权。" + exit 1 +fi + +# --------------------------------------------------------------------------- +# Step 2：直接用 authCode 换 token +# --------------------------------------------------------------------------- + +if [ -n "$1" ] && [ "$1" != "--check" ]; then + AUTH_CODE="$1" + echo "" + echo "🔄 正在用 authCode 换取 session token..." + RESPONSE=$(curl -s --max-time 20 \ + "$BACKEND/api/v1/auth/direct-token?authCode=${AUTH_CODE}") + + ERROR=$(echo "$RESPONSE" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('code',''))" 2>/dev/null) + if [ -n "$ERROR" ]; then + MSG=$(echo "$RESPONSE" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('msg',''))" 2>/dev/null) + echo "❌ 换取失败: [$ERROR] $MSG" + echo " 原始响应: $RESPONSE" + exit 1 + fi + + TOKEN=$(echo "$RESPONSE" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['token'])" 2>/dev/null) + USER=$(echo "$RESPONSE" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['user'].get('name','?'))" 2>/dev/null) + EXPIRES=$(echo "$RESPONSE" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['expires_at'])" 2>/dev/null) + + if [ -z "$TOKEN" ]; then + echo "❌ 响应解析失败，原始响应：" + echo "$RESPONSE" + exit 1 + fi + + echo "✅ 换取成功！" + echo " 用户：$USER" + echo " 有效期至：$EXPIRES" + save_token "$TOKEN" + echo "" + echo "🎉 认证完成，AI 现在可以访问 Holy Crab 后端了！" + exit 0 +fi + +# --------------------------------------------------------------------------- +# Step 3：交互式引导 — 获取 authCode +# --------------------------------------------------------------------------- + +echo "" +echo "═══════════════════════════════════════════════════════" +echo " 🔐 Holy Crab · AI 授权流程" +echo "═══════════════════════════════════════════════════════" +echo "" +echo " 本流程分两步：① 获取授权码 ② 换取 Token" +echo "" + +# 检查后端是否可达 +if ! curl -s --max-time 5 "$BACKEND/health" > /dev/null 2>&1; then + echo " ❌ 后端不可达: $BACKEND" + echo "" + echo " 请确认：" + echo " 1. 后端是否已启动？（python3 app.py）" + echo " 2. HC_BACKEND_URL 是否正确？" + echo "" + echo " 当前 BACKEND=$BACKEND" + echo " 修改方式：export HC_BACKEND_URL=http://服务器IP:8000" + exit 1 +fi +echo " ✅ 后端在线: $BACKEND" + +# 检查现有 token +if EXISTING=$(load_existing_token 2>/dev/null); then + if check_token "$EXISTING"; then + echo "" + echo " ✅ 已有有效 Token，无需重新授权。" + echo " Token 来源: ${ENV_FILE:-$COOKIE_FILE}" + exit 0 + else + echo " ⚠️ 已有 Token 但已过期，需要重新授权..." + fi +fi + +echo "" +echo "───────────────────────────────────────────────────────" +echo " Step 1：在浏览器中完成钉钉授权" +echo "───────────────────────────────────────────────────────" +echo "" + +# 获取钉钉授权 URL +AUTH_URL_RESP=$(curl -s --max-time 10 "$BACKEND/api/v1/auth/dingtalk/login?redirect=/") +AUTH_URL=$(echo "$AUTH_URL_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['url'])" 2>/dev/null) + +if [ -z "$AUTH_URL" ]; then + echo " ❌ 无法获取钉钉授权 URL（后端可能未配置 DINGTALK_CLIENT_ID）" + echo " 原始响应: $AUTH_URL_RESP" + exit 1 +fi + +echo " 请在浏览器中打开以下链接：" +echo "" +echo " $AUTH_URL" +echo "" +echo " 操作步骤：" +echo " 1. 浏览器会自动打开钉钉授权页面" +echo " 2. 点击「授权登录」" +echo " 3. 授权成功后，浏览器会跳转到后端页面" +echo " 4. 此时地址栏 URL 中会有 ?authCode=xxxxx" +echo " 把 authCode= 后面的那串字符复制下来" +echo "" + +# 自动打开浏览器 +if command -v open >/dev/null 2>&1; then + echo " 正在打开授权页面..." + open "$AUTH_URL" 2>/dev/null || true +fi + +echo "" +echo "───────────────────────────────────────────────────────" +echo " Step 2：把 authCode 贴过来" +echo "───────────────────────────────────────────────────────" +echo "" +echo " 复制 authCode 后，运行以下命令（把 YOUR_AUTH_CODE 换成你的值）：" +echo "" +echo " bash $0 YOUR_AUTH_CODE" +echo "" +echo " 或者直接告诉 AI：" +echo " 【authCode】YOUR_AUTH_CODE" +echo " AI 会自动帮你完成后续操作。" +echo "" +echo "═══════════════════════════════════════════════════════" diff --git a/skills/holy-crab/scripts/process_data.py b/skills/holy-crab/scripts/process_data.py new file mode 100755 index 0000000..0e18f46 --- /dev/null +++ b/skills/holy-crab/scripts/process_data.py @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +""" +process_data.py — 对 Holy Crab 任务结果进行数据分析 +用法: + python3 process_data.py summarize <任务ID> 榜单摘要 + python3 process_data.py sentiment <任务ID> 评论情感统计 + python3 process_data.py compare <任务ID> 产品横向对比 + python3 process_data.py detail <任务ID> 完整数据（含召回内容） + +Examples: + python3 process_data.py summarize HC-20260727-ABC12 + python3 process_data.py sentiment HC-20260727-ABC12 +""" + +import json +import os +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +DEFAULT_BACKEND = "http://localhost:8000" + + +def load_cookie() -> str | None: + env = os.environ.get("HC_SESSION") + if env: + return env + cookie_file = Path.home() / ".holy_crab_cookie" + if cookie_file.exists(): + return cookie_file.read_text().strip() + return None + + +def fetch_task(task_id: str) -> dict: + backend = os.environ.get("HC_BACKEND_URL", DEFAULT_BACKEND).rstrip("/") + url = f"{backend}/api/v1/tasks/{task_id}" + cookie = load_cookie() + headers = {} + if cookie: + headers["Cookie"] = f"hc_session={cookie}" + + import urllib.request + req = urllib.request.Request(url, headers=headers) + try: + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + try: + err = json.loads(body) + print(f"❌ API错误 {exc.code}: {err}", file=sys.stderr) + except Exception: + print(f"❌ HTTP {exc.code}: {body[:300]}", file=sys.stderr) + sys.exit(1) + + result = data.get("data", {}).get("result", {}) + extracted = result.get("extracted") + if not extracted: + print(f"❌ 任务 {task_id} 暂无 extracted 数据（任务未完成或格式不兼容）", + file=sys.stderr) + sys.exit(1) + return extracted + + +# --------------------------------------------------------------------------- +# 命令处理器 +# --------------------------------------------------------------------------- + +def cmd_summarize(extracted: dict) -> None: + """榜单摘要""" + ranking = extracted.get("榜单", []) + products = extracted.get("产品详情", []) + src = extracted.get("参考来源笔记总量", {}) + meta = extracted.get("提取元数据", {}) + stats = meta.get("召回统计", {}) + warns = extracted.get("提取警告", []) + + print(f"\n{'='*60}") + print(f" 📊 榜单摘要") + print(f"{'='*60}") + print(f" 参考来源笔记总量 : {src.get('原始文本', 'N/A')} (提取: {src.get('提取数量', 'N/A')})") + print(f" 包含召回明细 : {'是' if meta.get('包含召回明细') else '否'}") + print(f" 标签总数 : {stats.get('标签总数', 0)}") + print(f" 实际召回总数 : {stats.get('实际召回内容总数', 0)}") + print(f"\n 【榜单排名】") + print(f" {'排名':<4} {'产品名称':<20} {'推荐比例':<8} {'产品详情'}") + print(f" {'─'*60}") + for item in ranking: + rank = item.get("当前排名", "") + name = item.get("产品名称", "")[:20] + ratio = item.get("推荐比例", "") + # 找对应的产品详情 + prod = next((p for p in products if p.get("产品名称") == name), None) + if prod: + top3 = [t["内容标签"] for t in prod.get("标签数量Top3", [])] + detail = f"Top标签: {', '.join(top3)}" + else: + detail = "" + print(f" {str(rank):<4} {name:<20} {ratio:<8} {detail}") + + if products: + print(f"\n 【各产品经验标签统计】") + for prod in products: + tags = prod.get("内容标签", []) + top3 = prod.get("标签数量Top3", []) + print(f"\n ▶ {prod.get('产品名称')} (排行{prod.get('当前排行')})") + print(f" 参考经验: {prod.get('参考经验人数')} | 推荐比例: {prod.get('推荐比例')}") + print(f" 标签 Top3: ", end="") + print(", ".join(f"{t['内容标签']}({t['经验数量']})" for t in top3) if top3 else "无") + recall_total = sum(t.get("实际召回数量", 0) for t in tags) + print(f" 实际召回总数: {recall_total}") + + if warns: + print(f"\n ⚠️ 提取警告 ({len(warns)} 条):") + for w in warns[:5]: + print(f" - {w}") + if len(warns) > 5: + print(f" ... 共 {len(warns)} 条") + + print(f"\n{'='*60}\n") + + +def cmd_sentiment(extracted: dict) -> None: + """评论情感统计（从 comment 类型召回内容中提取）""" + all_comments = [] + for prod in extracted.get("产品详情", []): + for tag in prod.get("内容标签", []): + for note in tag.get("召回内容", []): + if note.get("召回类型") == "comment": + all_comments.append({ + "评论内容": note.get("评论内容"), + "所属笔记点赞": note.get("所属笔记点赞数量"), + "发布时间": note.get("发布时间"), + "产品": prod.get("产品名称"), + "标签": tag.get("内容标签"), + }) + + if not all_comments: + print(f"\n ⚠️ 当前任务无评论召回数据（comment 类型召回内容为空）") + print(f" 轻度/常规任务默认不包含评论召回，深度任务（deep模式）才有。\n") + return + + print(f"\n{'='*60}") + print(f" 💬 评论召回统计 (共 {len(all_comments)} 条)") + print(f"{'='*60}") + + # 按产品分组 + by_product: dict[str, list] = {} + for c in all_comments: + p = c["产品"] + by_product.setdefault(p, []).append(c) + + for product, comments in by_product.items(): + print(f"\n 【{product}】({len(comments)} 条评论)") + for c in comments[:10]: + content = c["评论内容"] or "" + print(f" • {content[:60]}{'...' if len(content)>60 else ''}") + print(f" 所属笔记 👍{c['所属笔记点赞']} | {c['发布时间']}") + if len(comments) > 10: + print(f" ... 还有 {len(comments)-10} 条") + + print(f"\n 📌 AI 情感分析建议：") + print(f" 请 AI 读取上述评论内容，判断每条评论的情感（正面/负面/中性），") + print(f" 并统计各情感类别的数量和占比。") + print(f"{'='*60}\n") + + +def cmd_compare(extracted: dict) -> None: + """产品横向对比""" + products = extracted.get("产品详情", []) + + if not products: + print(f"❌ 无产品详情数据", file=sys.stderr) + return + + print(f"\n{'='*60}") + print(f" 🔍 产品横向对比") + print(f"{'='*60}") + + headers = ["指标"] + [p.get("产品名称", f"产品{i+1}")[:12] for i, p in enumerate(products)] + col_w = 14 + print(f" {'指标':<14} " + " ".join(f"{h:<{col_w}}" for h in headers[1:])) + print(f" {'─'*60}") + + def col(val: str) -> str: + return f"{str(val):<{col_w}}" + + print(f" {'当前排行':<14} " + " ".join(col(p.get("当前排行", "")) for p in products)) + print(f" {'参考经验人数':<14} " + " ".join(col(p.get("参考经验人数", "")) for p in products)) + print(f" {'推荐比例':<14} " + " ".join(col(p.get("推荐比例", "")) for p in products)) + + tag_counts = [len(p.get("内容标签", [])) for p in products] + print(f" {'标签数量':<14} " + " ".join(col(c) for c in tag_counts)) + + recall_totals = [] + for p in products: + total = sum(t.get("实际召回数量", 0) for t in p.get("内容标签", [])) + recall_totals.append(total) + print(f" {'实际召回总数':<14} " + " ".join(col(r) for r in recall_totals)) + + top_tags = [] + for p in products: + top3 = p.get("标签数量Top3", []) + tags_str = ", ".join(t["内容标签"] for t in top3[:2]) if top3 else "—" + top_tags.append(tags_str) + print(f" {'Top1-2标签':<14} " + " ".join(f"{t[:col_w*2-2]:<{col_w*2}}" for t in top_tags)) + + print(f"\n 【各产品 Top3 标签详情】") + for i, prod in enumerate(products): + print(f"\n {i+1}. {prod.get('产品名称')}") + for t in prod.get("标签数量Top3", []): + recall = next( + (x.get("实际召回数量", 0) + for x in prod.get("内容标签", []) + if x.get("内容标签") == t["内容标签"]), + 0 + ) + print(f" {t['排名']}. {t['内容标签']} {t['经验数量']}人 召回:{recall}") + + print(f"\n{'='*60}\n") + + +def cmd_detail(extracted: dict) -> None: + """完整数据（含所有召回内容）""" + print(json.dumps(extracted, ensure_ascii=False, indent=2)) + + +# --------------------------------------------------------------------------- +# 主入口 +# --------------------------------------------------------------------------- + +COMMANDS = { + "summarize": ("榜单摘要", cmd_summarize), + "sentiment": ("评论情感统计", cmd_sentiment), + "compare": ("产品横向对比", cmd_compare), + "detail": ("完整数据", cmd_detail), +} + + +def main() -> None: + if len(sys.argv) < 3 or sys.argv[1] not in COMMANDS: + print(f"用法: {sys.argv[0]} <{'|'.join(COMMANDS)}> <任务ID>") + print(f"示例: {sys.argv[0]} summarize HC-20260727-ABC12") + sys.exit(1) + + cmd_key = sys.argv[1] + task_id = sys.argv[2] + _, handler = COMMANDS[cmd_key] + + extracted = fetch_task(task_id) + handler(extracted) + + +if __name__ == "__main__": + main() diff --git a/skills/holy-crab/scripts/query_tasks.py b/skills/holy-crab/scripts/query_tasks.py new file mode 100755 index 0000000..09db808 --- /dev/null +++ b/skills/holy-crab/scripts/query_tasks.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +""" +query_tasks.py — 列出 Holy Crab 所有任务 +用法: python3 query_tasks.py [任务ID 关键词] + +Examples: + python3 query_tasks.py + python3 query_tasks.py HC-20260727-ABC12 +""" + +import json +import os +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +DEFAULT_BACKEND = "http://localhost:8000" + + +def load_cookie() -> str | None: + # 优先级：环境变量 > ~/.holy_crab_env > ~/.holy_crab_cookie + env = os.environ.get("HC_SESSION") + if env: + return env + env_file = Path.home() / ".holy_crab_env" + if env_file.exists(): + for line in env_file.read_text().strip().splitlines(): + line = line.strip() + if line.startswith("HC_SESSION=") or line.startswith("hc_session="): + val = line.split("=", 1)[1].strip().strip('"').strip("'") + if val: + return val + cookie_file = Path.home() / ".holy_crab_cookie" + if cookie_file.exists(): + return cookie_file.read_text().strip() + return None + + +def api_get(path: str, params: dict | None = None) -> dict: + backend = os.environ.get("HC_BACKEND_URL", DEFAULT_BACKEND).rstrip("/") + url = f"{backend}/api/v1{path}" + cookie = load_cookie() + headers = {} + if cookie: + headers["Cookie"] = f"hc_session={cookie}" + + import urllib.request + req = urllib.request.Request(url, headers=headers) + if params: + import urllib.parse + req.full_url = f"{url}?{urllib.parse.urlencode(params)}" + try: + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + try: + err = json.loads(body) + print(f"❌ API错误 {exc.code}: {err}", file=sys.stderr) + except Exception: + print(f"❌ HTTP {exc.code}: {body[:300]}", file=sys.stderr) + sys.exit(1) + + +def format_status(status: str) -> str: + mapping = { + "completed": "✅ 完成", + "failed": "❌ 失败", + "cancelled": "🚫 取消", + "partial": "⚠️ 部分", + "running": "⏳ 进行中", + "waiting": "⏳ 等待中", + } + return mapping.get(status, status) + + +def main() -> None: + task_id_filter = sys.argv[1] if len(sys.argv) > 1 else None + + if task_id_filter: + # 精确查单个任务 + data = api_get(f"/tasks/{task_id_filter}") + task = data.get("data", {}) + print(f"\n{'─'*60}") + print(f" 任务ID : {task.get('task_id')}") + print(f" 名称 : {task.get('name')}") + print(f" 关键词 : {', '.join(task.get('keywords', []))}") + print(f" 状态 : {format_status(task.get('status'))}") + print(f" 进度 : {task.get('progress_percent')}%") + print(f" 模式 : {task.get('mode')}") + print(f" 创建时间 : {task.get('created_at')}") + print(f" 完成时间 : {task.get('completed_at') or '—'}") + if task.get("error"): + print(f" 错误 : [{task['error'].get('code')}] {task['error'].get('message')}") + # 检查 extracted + result = task.get("result", {}) + extracted = result.get("extracted") + if extracted: + meta = extracted.get("提取元数据", {}) + stats = meta.get("召回统计", {}) + print(f"\n 📊 数据清洗结果:") + print(f" schema : {extracted.get('schema_version')}") + print(f" 标签总数 : {stats.get('标签总数', 'N/A')}") + print(f" 召回总数 : {stats.get('实际召回内容总数', 'N/A')}") + print(f" 榜单数量 : {len(extracted.get('榜单', []))}") + print(f" 产品数量 : {len(extracted.get('产品详情', []))}") + src = extracted.get("参考来源笔记总量", {}) + print(f" 笔记总量 : {src.get('原始文本', 'N/A')}") + else: + print(f"\n ⏳ 任务尚未完成或无 extracted 数据") + print(f"{'─'*60}\n") + else: + # 列表 + data = api_get("/tasks", {"page_size": 50}) + items = data.get("data", {}).get("items", []) + total = data.get("data", {}).get("total", 0) + print(f"\n{'─'*60}") + print(f" Holy Crab 任务列表 (共 {total} 个)") + print(f"{'─'*60}") + print(f" {'任务ID':<22} {'状态':<14} {'关键词':<16} {'创建时间':<12} {'文件'}") + print(f" {'─'*60}") + for t in items: + keywords = ",".join(t.get("keywords", [])[:2]) + if len(",".join(t.get("keywords", []))) > 16: + keywords += "..." + status = format_status(t.get("status", "")) + files = len(t.get("files", [])) + ts = (t.get("created_at") or "")[:10] + print(f" {t.get('task_id'):<22} {status:<14} {keywords:<16} {ts:<12} {files}f") + print(f"{'─'*60}\n") + + +if __name__ == "__main__": + main() diff --git a/task_service.py b/task_service.py new file mode 100644 index 0000000..69b4f98 --- /dev/null +++ b/task_service.py @@ -0,0 +1,648 @@ +import hashlib +import json +import secrets +import string +import threading +import time +from datetime import datetime +from pathlib import Path +from typing import Any + +from config import Config +from crawler_client import ( + CrawlerError, + cancel_collection, + query_collection, + resume_collection, + start_collection, +) +from database import connection, json_dumps, json_loads, now_iso +from extract_service import transform_result + + +MODE_NAMES = {"mild": "轻度", "regular": "常规", "deep": "深度"} +TERMINAL_STATUSES = {"completed", "failed", "cancelled", "partial"} +# 远程 overview/sources 初始化较重，同一业务进程内串行提交， +# 避免多个新任务同时触发问一问基础采集导致网关超时。 +COLLECTION_START_LOCK = threading.Lock() + + +class TaskError(RuntimeError): + def __init__(self, code: str, message: str, status_code: int = 400, details: dict | None = None): + super().__init__(message) + self.code = code + self.message = message + self.status_code = status_code + self.details = details or {} + + +def _public_task_id() -> str: + day = datetime.now().strftime("%Y%m%d") + chars = string.ascii_uppercase + string.digits + return f"HC-{day}-{''.join(secrets.choice(chars) for _ in range(5))}" + + +def _file_id() -> str: + return f"file_{secrets.token_hex(8)}" + + +def validate_create(data: dict[str, Any]) -> tuple[str, str, list[str], str, str]: + mode = str(data.get("mode") or "") + if mode not in MODE_NAMES: + raise TaskError("INVALID_MODE", "mode 必须是 mild、regular 或 deep", 422) + platform = str(data.get("platform") or "") + if platform != "xiaohongshu_wenyiwen": + raise TaskError("INVALID_PLATFORM", "第一期仅支持 xiaohongshu_wenyiwen", 422) + raw_keywords = data.get("keywords") + if not isinstance(raw_keywords, list): + raise TaskError("KEYWORD_REQUIRED", "keywords 必须是数组", 422) + keywords = list(dict.fromkeys(str(item).strip() for item in raw_keywords if str(item).strip())) + if not keywords: + raise TaskError("KEYWORD_REQUIRED", "至少需要一个搜索词", 422) + if mode == "mild" and len(keywords) > 50: + raise TaskError("KEYWORD_LIMIT_EXCEEDED", "轻度任务最多支持 50 个搜索词", 422) + if mode in {"regular", "deep"} and len(keywords) != 1: + raise TaskError("SINGLE_KEYWORD_REQUIRED", "常规和深度任务只能包含一个搜索词", 422) + if any(len(keyword) > 200 for keyword in keywords): + raise TaskError("INVALID_KEYWORD", "单个搜索词不能超过 200 个字符", 422) + name = str(data.get("name") or "").strip() or f"{MODE_NAMES[mode]}｜问一问｜{keywords[0]}" + if len(name) > 100: + raise TaskError("INVALID_TASK_NAME", "任务名称不能超过 100 个字符", 422) + created_by_id = str(data.get("created_by_id") or "usr_local") + created_by_name = str(data.get("created_by_name") or "本地用户") + return mode, platform, keywords, name, created_by_id, created_by_name + + +def create_task(data: dict[str, Any], retry_of_task_id: int | None = None) -> dict[str, Any]: + mode, platform, keywords, name, creator_id, creator_name = validate_create(data) + now = now_iso() + with connection() as conn: + while True: + public_id = _public_task_id() + if not conn.execute("SELECT 1 FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone(): + break + cursor = conn.execute( + """INSERT INTO research_tasks + (public_task_id,name,platform,mode,status,execution_stage,progress_percent, + created_by_id,created_by_name,retry_of_task_id,created_at,updated_at) + VALUES (?,?,?,?, 'waiting','queued',0,?,?,?,?,?)""", + (public_id, name, platform, mode, creator_id, creator_name, retry_of_task_id, now, now), + ) + task_pk = cursor.lastrowid + conn.executemany( + """INSERT INTO task_keywords + (task_id,keyword_order,keyword,normalized_keyword,created_at) + VALUES (?,?,?,?,?)""", + [(task_pk, index, keyword, keyword.casefold(), now) for index, keyword in enumerate(keywords)], + ) + _event(conn, task_pk, "created", "queued", 0, "user", creator_id, {"keywords": keywords}) + task = get_task(public_id) + threading.Thread(target=_execute_task, args=(public_id,), daemon=True, name=f"holy-crab-{public_id}").start() + return task + + +def _event(conn, task_pk: int, event_type: str, stage: str, progress: int, actor_type: str, + actor_id: str | None = None, payload: dict | None = None) -> None: + conn.execute( + """INSERT INTO task_events + (task_id,event_type,stage,progress_percent,actor_type,actor_id,payload_json,occurred_at) + VALUES (?,?,?,?,?,?,?,?)""", + (task_pk, event_type, stage, progress, actor_type, actor_id, json_dumps(payload) if payload else None, now_iso()), + ) + + +def _keywords(conn, task_pk: int) -> list[str]: + rows = conn.execute( + "SELECT keyword FROM task_keywords WHERE task_id=? ORDER BY keyword_order", (task_pk,) + ).fetchall() + return [row["keyword"] for row in rows] + + +def _update(public_id: str, *, status: str | None = None, stage: str | None = None, + progress: int | float | None = None, crawler_task_id: str | None = None, + result: dict | None = None, error_code: str | None = None, + error_message: str | None = None, completed: bool = False) -> None: + with connection() as conn: + row = conn.execute("SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row: + return + # cancelled 任务仍允许写入最终部分结果，但禁止恢复为其他状态。 + if row["status"] == "cancelled" and result is None: + return + next_status = "cancelled" if row["status"] == "cancelled" else (status or row["status"]) + next_progress = max(row["progress_percent"], min(100, int(float(progress or 0)))) + now = now_iso() + conn.execute( + """UPDATE research_tasks SET status=?,execution_stage=?,progress_percent=?, + crawler_task_id=COALESCE(?,crawler_task_id), + result_json=COALESCE(?,result_json),error_code=?,error_message=?, + started_at=COALESCE(started_at,?), + completed_at=CASE WHEN ? THEN ? ELSE completed_at END, + updated_at=?,version=version+1 WHERE id=?""", + ( + next_status, stage or row["execution_stage"], next_progress, + crawler_task_id, json_dumps(result) if result is not None else None, + error_code, error_message, now, completed, now, now, row["id"], + ), + ) + _event(conn, row["id"], "status_changed", stage or row["execution_stage"], next_progress, + "worker", payload={"status": next_status, "error_code": error_code}) + + +def _execute_task(public_id: str) -> None: + try: + with connection() as conn: + row = conn.execute("SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row or row["status"] == "cancelled": + return + keywords = _keywords(conn, row["id"]) + mode = row["mode"] + _update(public_id, status="running", stage="initializing", progress=5) + with COLLECTION_START_LOCK: + result = start_collection(mode, keywords) + crawler_task_id = str(result.get("task_id") or "") + if not crawler_task_id: + raise CrawlerError("爬虫服务未返回 task_id") + with connection() as conn: + local = conn.execute( + "SELECT status FROM research_tasks WHERE public_task_id=?", (public_id,) + ).fetchone() + if local and local["status"] == "cancelled": + cancel_collection(crawler_task_id) + return + + crawler_status = str(result.get("status") or "") + is_async = mode == "deep" or crawler_status in {"running", "waiting"} + if not is_async: + _complete_task(public_id, result, crawler_task_id=crawler_task_id) + return + + _update( + public_id, status="running", stage="collecting_search", + progress=max(10, float(result.get("progress") or 10)), + crawler_task_id=crawler_task_id, result=result, + ) + while True: + time.sleep(Config.CRAWLER_POLL_INTERVAL_SECONDS) + with connection() as conn: + local = conn.execute( + "SELECT status FROM research_tasks WHERE public_task_id=?", (public_id,) + ).fetchone() + if not local or local["status"] == "cancelled": + return + queried = query_collection(crawler_task_id) + status = str(queried.get("status") or "running") + if status in {"completed", "partial"}: + _complete_task( + public_id, queried, status=status, + crawler_task_id=crawler_task_id, + ) + return + if status == "failed": + raise CrawlerError(str(queried.get("error_message") or "爬虫任务执行失败")) + _update( + public_id, status="running", + stage=str(queried.get("current_stage") or "collecting_search"), + progress=float(queried.get("progress") or 10), + crawler_task_id=crawler_task_id, result=queried, + ) + except Exception as exc: + _update( + public_id, status="failed", stage="failed", progress=0, + error_code="COLLECTOR_FAILED", error_message=str(exc), completed=True, + ) + + +def _complete_task( + public_id: str, + result: dict[str, Any], + status: str = "completed", + crawler_task_id: str | None = None, +) -> None: + # 对原始爬虫结果执行数据清洗，生成结构化的 extracted 字段 + result = transform_result(result) + _write_result_files(public_id, result) + _update( + public_id, status=status, stage="completed", progress=100, + crawler_task_id=crawler_task_id, result=result, completed=True, + error_code=("PARTIAL_RESULT" if status == "partial" else None), + error_message=("部分数据采集失败" if status == "partial" else None), + ) + + +def _write_result_files(public_id: str, result: dict[str, Any]) -> None: + folder = Config.FILE_STORAGE_PATH / public_id + folder.mkdir(parents=True, exist_ok=True) + json_path = folder / "xhs_wen_result.json" + summary_path = folder / "research_summary.md" + json_bytes = json.dumps(result, ensure_ascii=False, indent=2).encode("utf-8") + json_path.write_bytes(json_bytes) + summary = ( + f"# 问一问任务数据摘要\n\n" + f"- 任务 ID：{public_id}\n" + f"- 生成时间：{now_iso()}\n" + f"- 数据文件：xhs_wen_result.json\n" + f"- SHA-256：{hashlib.sha256(json_bytes).hexdigest()}\n" + ) + summary_path.write_text(summary, encoding="utf-8") + with connection() as conn: + row = conn.execute("SELECT id FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row: + return + for path, stage, category, fmt, count in [ + (json_path, "cleaned", "search_result", "json", _result_count(result)), + (summary_path, "cleaned", "metadata", "md", 1), + ]: + content = path.read_bytes() + existing = conn.execute( + "SELECT public_file_id FROM task_files WHERE task_id=? AND name=?", + (row["id"], path.name), + ).fetchone() + public_file_id = existing["public_file_id"] if existing else _file_id() + conn.execute( + """INSERT OR REPLACE INTO task_files + (public_file_id,task_id,name,stage,category,format,record_count,size_bytes, + storage_key,sha256,schema_version,generated_at,available) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + public_file_id, row["id"], path.name, stage, category, fmt, + count, len(content), str(path), hashlib.sha256(content).hexdigest(), + "1.0", now_iso(), 1, + ), + ) + + +def _result_count(result: dict[str, Any]) -> int: + data = result.get("data") + if isinstance(data, list): + return len(data) + items = result.get("items") + return len(items) if isinstance(items, list) else 1 + + +def _task_dict(conn, row, include_result: bool = True) -> dict[str, Any]: + keywords = _keywords(conn, row["id"]) + saved_result = json_loads(row["result_json"]) if row["result_json"] else None + # 兼容修复前生成的同步任务：爬虫任务 ID 曾只保存在 result.task_id。 + collector_task_id = row["crawler_task_id"] + if not collector_task_id and isinstance(saved_result, dict): + collector_task_id = saved_result.get("task_id") + files = conn.execute( + "SELECT * FROM task_files WHERE task_id=? AND available=1 ORDER BY id", (row["id"],) + ).fetchall() + data = { + "task_id": row["public_task_id"], + "collector_task_id": collector_task_id, + "name": row["name"], + "mode": row["mode"], + "platform": {"code": row["platform"], "name": "问一问"}, + "keywords": keywords, + "status": row["status"], + "execution_stage": row["execution_stage"], + "progress_percent": row["progress_percent"], + "created_by": {"id": row["created_by_id"], "name": row["created_by_name"]}, + "created_at": row["created_at"], + "updated_at": row["updated_at"], + "started_at": row["started_at"], + "completed_at": row["completed_at"], + "retry_of_task_id": _retry_public_id(conn, row["retry_of_task_id"]), + "cancel_requested": bool(row["cancel_requested_at"]), + "error": ( + {"code": row["error_code"], "message": row["error_message"]} + if row["error_code"] or row["error_message"] else None + ), + "usable_by_agent": row["status"] == "completed", + "files": [_file_dict(file) for file in files], + } + if include_result and saved_result is not None: + # 兜底：对历史任务（未经过 transform_result 的旧数据） + # 也执行清洗转换，确保前端 / AI 始终拿到带 extracted 的完整结果 + data["result"] = transform_result(saved_result) + return data + + +def _retry_public_id(conn, retry_pk: int | None) -> str | None: + if not retry_pk: + return None + row = conn.execute("SELECT public_task_id FROM research_tasks WHERE id=?", (retry_pk,)).fetchone() + return row["public_task_id"] if row else None + + +def _file_dict(row) -> dict[str, Any]: + return { + "file_id": row["public_file_id"], "name": row["name"], "format": row["format"], + "stage": row["stage"], "stage_name": "清洗数据" if row["stage"] == "cleaned" else "原始数据", + "category": row["category"], "category_name": { + "search_result": "搜索结果", "entry": "收录内容", + "comment": "评论", "metadata": "任务元数据", + }.get(row["category"], row["category"]), + "record_count": row["record_count"], "size_bytes": row["size_bytes"], + "sha256": row["sha256"], "schema_version": row["schema_version"], + "generated_at": row["generated_at"], "previewable": row["size_bytes"] <= 2 * 1024 * 1024, + } + + +def get_task(public_id: str) -> dict[str, Any]: + with connection() as conn: + row = conn.execute("SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row: + raise TaskError("TASK_NOT_FOUND", "任务不存在", 404) + return _task_dict(conn, row) + + +def get_task_extracted(public_id: str) -> dict[str, Any]: + """直接返回任务的清洗解析结果，方便 curl / Agent 读取。 + + 普通任务详情接口需要服务前端展示，因此会包装任务状态、文件列表和原始结果。 + Agent 做分析时更适合直接读取 result.extracted，这里把 extracted 提升到 + data 顶层返回。 + """ + task = get_task(public_id) + result = task.get("result") if isinstance(task, dict) else None + extracted = result.get("extracted") if isinstance(result, dict) else None + if not isinstance(extracted, dict): + raise TaskError("EXTRACTED_NOT_READY", "任务清洗数据尚未生成", 404) + extracted = dict(extracted) + metadata = dict(extracted.get("提取元数据") or {}) + metadata.setdefault("来源任务ID", public_id) + metadata.setdefault("核心数据路径", "data.result.extracted") + metadata.setdefault("采集任务ID", task.get("collector_task_id")) + metadata.setdefault("任务状态", task.get("status")) + metadata.setdefault("关键词", task.get("keywords")) + extracted["提取元数据"] = metadata + return extracted + + +def list_tasks(q: str = "", status: str = "", page: int = 1, page_size: int = 10) -> dict[str, Any]: + if page < 1 or page_size not in {10, 20, 50}: + raise TaskError("INVALID_PAGINATION", "page 必须大于 0，page_size 只允许 10、20、50", 422) + clauses, params = [], [] + if status: + clauses.append("t.status=?") + params.append(status) + if q: + clauses.append( + "(t.name LIKE ? OR t.public_task_id LIKE ? OR EXISTS " + "(SELECT 1 FROM task_keywords k WHERE k.task_id=t.id AND k.keyword LIKE ?))" + ) + like = f"%{q}%" + params.extend([like, like, like]) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + with connection() as conn: + total = conn.execute(f"SELECT COUNT(*) AS count FROM research_tasks t {where}", params).fetchone()["count"] + rows = conn.execute( + f"SELECT t.* FROM research_tasks t {where} ORDER BY t.created_at DESC LIMIT ? OFFSET ?", + [*params, page_size, (page - 1) * page_size], + ).fetchall() + return { + "items": [_task_dict(conn, row, include_result=False) for row in rows], + "page": page, "page_size": page_size, "total": total, + "total_pages": max(1, (total + page_size - 1) // page_size), + } + + +def cancel_task(public_id: str, reason: str = "") -> dict[str, Any]: + crawler_task_id = None + with connection() as conn: + row = conn.execute("SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row: + raise TaskError("TASK_NOT_FOUND", "任务不存在", 404) + if row["status"] == "cancelled": + return _task_dict(conn, row) + if row["status"] not in {"waiting", "running"}: + raise TaskError("TASK_STATUS_CONFLICT", "当前任务状态不允许取消", 409, {"current_status": row["status"]}) + crawler_task_id = row["crawler_task_id"] + if not crawler_task_id and row["result_json"]: + saved_result = json_loads(row["result_json"]) + if isinstance(saved_result, dict): + crawler_task_id = saved_result.get("task_id") + if crawler_task_id: + try: + cancel_collection(str(crawler_task_id)) + except CrawlerError as exc: + raise TaskError("COLLECTOR_UNAVAILABLE", f"爬虫任务停止失败：{exc}", 503) from exc + with connection() as conn: + row = conn.execute("SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row: + raise TaskError("TASK_NOT_FOUND", "任务不存在", 404) + if row["status"] not in {"waiting", "running", "cancelled"}: + raise TaskError("TASK_STATUS_CONFLICT", "任务状态已变化，无法停止", 409) + now = now_iso() + conn.execute( + """UPDATE research_tasks SET status='cancelled',execution_stage='cancelled', + cancel_requested_at=?,completed_at=?,updated_at=?,version=version+1 WHERE id=?""", + (now, now, now, row["id"]), + ) + _event(conn, row["id"], "cancelled", "cancelled", row["progress_percent"], "user", + payload={"reason": reason, "crawler_task_id": crawler_task_id}) + if crawler_task_id: + threading.Thread( + target=_sync_cancelled_result, + args=(public_id, str(crawler_task_id)), + daemon=True, + name=f"holy-crab-cancel-{public_id}", + ).start() + return get_task(public_id) + + +def _sync_cancelled_result(public_id: str, crawler_task_id: str) -> None: + """等待爬虫保存停止快照，并同步到业务任务结果和文件。""" + + deadline = time.time() + max(60, Config.CRAWLER_TIMEOUT_SECONDS) + latest = None + while time.time() < deadline: + try: + latest = query_collection(crawler_task_id) + except CrawlerError: + time.sleep(Config.CRAWLER_POLL_INTERVAL_SECONDS) + continue + if ( + latest.get("status") == "cancelled" + and latest.get("current_stage") == "cancelled" + and latest.get("data") is not None + ): + break + time.sleep(Config.CRAWLER_POLL_INTERVAL_SECONDS) + if latest and latest.get("data") is not None: + _write_result_files(public_id, latest) + _update( + public_id, status="cancelled", stage="cancelled", + progress=float(latest.get("progress") or 0), + crawler_task_id=crawler_task_id, result=latest, completed=True, + error_code="TASK_CANCELLED", + error_message="任务已停止，返回停止前已采集数据", + ) + + +def resume_task(public_id: str) -> dict[str, Any]: + """继续原业务任务：有爬虫任务 ID 时断点恢复，否则从初始化阶段重试。""" + with connection() as conn: + row = conn.execute( + "SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,) + ).fetchone() + if not row: + raise TaskError("TASK_NOT_FOUND", "任务不存在", 404) + if row["status"] not in {"failed", "cancelled", "partial"}: + raise TaskError( + "TASK_STATUS_CONFLICT", + "只有失败、部分完成或已取消任务可以继续采集", + 409, + ) + crawler_task_id = row["crawler_task_id"] + if not crawler_task_id and row["result_json"]: + saved_result = json_loads(row["result_json"]) + if isinstance(saved_result, dict): + crawler_task_id = saved_result.get("task_id") + if not crawler_task_id: + # 首次请求爬虫服务就失败时不会产生爬虫任务 ID，此时仍沿用 + # 原业务任务 ID，从初始化阶段重试，避免“继续采集”创建新任务。 + now = now_iso() + conn.execute( + """UPDATE research_tasks + SET status='waiting',execution_stage='queued',progress_percent=0, + cancel_requested_at=NULL,completed_at=NULL,error_code=NULL, + error_message=NULL,result_json=NULL,updated_at=?,version=version+1 + WHERE id=?""", + (now, row["id"]), + ) + _event( + conn, row["id"], "retried", "queued", 0, "user", + payload={"resume_mode": "restart_initialization"}, + ) + + if not crawler_task_id: + threading.Thread( + target=_execute_task, + args=(public_id,), + daemon=True, + name=f"holy-crab-retry-{public_id}", + ).start() + return get_task(public_id) + + try: + resumed = resume_collection(str(crawler_task_id)) + except CrawlerError as exc: + raise TaskError( + "COLLECTOR_UNAVAILABLE", f"爬虫任务恢复失败：{exc}", 503 + ) from exc + + with connection() as conn: + row = conn.execute( + "SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,) + ).fetchone() + if not row: + raise TaskError("TASK_NOT_FOUND", "任务不存在", 404) + now = now_iso() + progress = max(0, min(99, int(float(resumed.get("progress") or row["progress_percent"] or 0)))) + conn.execute( + """UPDATE research_tasks + SET status='running',execution_stage='resuming',progress_percent=?, + cancel_requested_at=NULL,completed_at=NULL,error_code=NULL, + error_message=NULL,result_json=?,updated_at=?,version=version+1 + WHERE id=?""", + (progress, json_dumps(resumed), now, row["id"]), + ) + _event( + conn, row["id"], "resumed", "resuming", progress, "user", + payload={"crawler_task_id": crawler_task_id}, + ) + + threading.Thread( + target=_monitor_resumed_task, + args=(public_id, str(crawler_task_id)), + daemon=True, + name=f"holy-crab-resume-{public_id}", + ).start() + return get_task(public_id) + + +def _monitor_resumed_task(public_id: str, crawler_task_id: str) -> None: + """轮询恢复后的原爬虫任务，并覆盖原业务任务的结果文件。""" + try: + while True: + time.sleep(Config.CRAWLER_POLL_INTERVAL_SECONDS) + with connection() as conn: + local = conn.execute( + "SELECT status FROM research_tasks WHERE public_task_id=?", + (public_id,), + ).fetchone() + if not local or local["status"] == "cancelled": + return + queried = query_collection(crawler_task_id) + status = str(queried.get("status") or "running") + if status in {"completed", "partial"}: + _complete_task( + public_id, queried, status=status, + crawler_task_id=crawler_task_id, + ) + return + if status == "failed": + raise CrawlerError( + str(queried.get("error_message") or "恢复采集失败") + ) + _update( + public_id, + status="running", + stage=str(queried.get("current_stage") or "resuming"), + progress=float(queried.get("progress") or 0), + crawler_task_id=crawler_task_id, + result=queried, + ) + except Exception as exc: + _update( + public_id, + status="failed", + stage="resume_failed", + progress=0, + error_code="COLLECTOR_RESUME_FAILED", + error_message=str(exc), + completed=True, + ) + + +def recollect_task(public_id: str) -> dict[str, Any]: + """复制原配置并创建全新的业务任务和爬虫任务，从头采集。""" + with connection() as conn: + row = conn.execute("SELECT * FROM research_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not row: + raise TaskError("TASK_NOT_FOUND", "任务不存在", 404) + if row["status"] in {"waiting", "running"}: + raise TaskError("TASK_STATUS_CONFLICT", "运行中的任务不能重新采集", 409) + keywords = _keywords(conn, row["id"]) + data = { + "name": row["name"], "mode": row["mode"], "platform": row["platform"], + "keywords": keywords, "created_by_id": row["created_by_id"], + "created_by_name": row["created_by_name"], + } + source_pk = row["id"] + return create_task(data, retry_of_task_id=source_pk) + + +# 保留旧函数名供其他内部调用兼容；语义已经调整为断点恢复。 +retry_task = resume_task + + +def get_file(public_id: str, file_id: str) -> tuple[dict[str, Any], Path]: + with connection() as conn: + row = conn.execute( + """SELECT f.* FROM task_files f JOIN research_tasks t ON t.id=f.task_id + WHERE t.public_task_id=? AND f.public_file_id=? AND f.available=1""", + (public_id, file_id), + ).fetchone() + if not row: + raise TaskError("FILE_NOT_FOUND", "文件不存在", 404) + return _file_dict(row), Path(row["storage_key"]) + + +def resume_unfinished_tasks() -> None: + """服务重启后恢复等待任务；已有爬虫任务则继续查询，否则重新提交。""" + + with connection() as conn: + rows = conn.execute( + "SELECT public_task_id FROM research_tasks WHERE status IN ('waiting','running')" + ).fetchall() + for row in rows: + threading.Thread( + target=_execute_task, args=(row["public_task_id"],), daemon=True, + name=f"holy-crab-resume-{row['public_task_id']}", + ).start()