From 0a3477cc4464f027c765fd5bdc6065f2ef5846f9 Mon Sep 17 00:00:00 2001 From: meijiali <你的邮箱@xxx.com> Date: Fri, 3 Jul 2026 12:20:59 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E6=8A=A5=E5=91=8A?= =?UTF-8?q?=E6=91=98=E8=A6=81=E7=94=9F=E6=88=90=E5=B9=B6=E7=A8=B3=E5=AE=9A?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E7=AB=AF=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/platforms/douyin.py | 22 ++++++-- app/services/ai_service.py | 44 +++++++++++++++- app/services/report_service.py | 7 ++- app/services/task_service.py | 44 +++++++++++++--- docker-compose.yml | 2 + .../integration/test_task_flow_xiaohongshu.py | 27 ++++++++-- tests/unit/test_ai_schema.py | 52 +++++++++++++++++++ tests/unit/test_deployment_config.py | 6 +++ tests/unit/test_douyin_mapping.py | 36 +++++++++++++ tests/unit/test_task_executor.py | 10 ++++ 10 files changed, 230 insertions(+), 20 deletions(-) diff --git a/app/platforms/douyin.py b/app/platforms/douyin.py index 4e8f16f..684bc07 100644 --- a/app/platforms/douyin.py +++ b/app/platforms/douyin.py @@ -88,11 +88,23 @@ class DouyinPlatform: return self.map_items(payload, limit=limit) def fetch_comments(self, source_item_id: str, *, limit: int) -> list[CommentData]: - payload = self.client.get( - "/api/v1/douyin/app/v3/fetch_video_comments", - params={"aweme_id": source_item_id, "cursor": 0, "count": 20}, - ) - return self.map_comments(payload, limit=limit) + comments: list[CommentData] = [] + cursor = 0 + page_size = min(20, limit) + while len(comments) < limit: + payload = self.client.get( + "/api/v1/douyin/app/v3/fetch_video_comments", + params={"aweme_id": source_item_id, "cursor": cursor, "count": page_size}, + ) + page_comments = self.map_comments(payload, limit=limit - len(comments)) + comments.extend(page_comments) + data = payload.get("data", {}) + next_cursor = data.get("cursor") or payload.get("cursor") + has_more = data.get("has_more", payload.get("has_more", 0)) + if not page_comments or not has_more or next_cursor in (None, cursor): + break + cursor = next_cursor + return comments[:limit] def _author_name(self, comment: dict[str, Any]) -> str | None: user = comment.get("user") or {} diff --git a/app/services/ai_service.py b/app/services/ai_service.py index 06e0d19..f48bd97 100644 --- a/app/services/ai_service.py +++ b/app/services/ai_service.py @@ -42,6 +42,46 @@ def build_comment_prompt(comments: list[dict[str, str]]) -> str: ) +def build_report_summary_prompt(metrics: dict, typical: dict, *, word_limit: int) -> str: + sentiment = metrics.get("sentiment", {}) + sentiment_lines = [] + for name in ("positive", "neutral", "negative", "unknown"): + entry = sentiment.get(name, {}) + sentiment_lines.append(f"- {name}: {entry.get('count', 0)} ({entry.get('pct', 0)}%)") + + label_lines = [ + f"- {label.get('name', '')}: {label.get('count', 0)}" + for label in metrics.get("top_labels", []) + if label.get("name") + ] + if not label_lines: + label_lines = ["- 暂无"] + + typical_lines = [] + for sentiment_name, comments in typical.items(): + for comment in comments: + typical_lines.append(f"- [{sentiment_name}] {str(comment.get('content', ''))[:150]}") + if not typical_lines: + typical_lines = ["- 暂无"] + + item_count_line = "" + if "item_count" in metrics: + item_count_line = f"内容条目数量：{metrics.get('item_count', 0)}\n" + + return ( + f"只返回一段中文总结，不使用 Markdown，不超过 {word_limit} 字。\n" + "总结必须基于以下统计数据和典型评论，不要编造未提供的信息。\n" + f"{item_count_line}" + f"样本评论数量：{metrics.get('sample_count', 0)}\n" + "情绪分布：\n" + f"{chr(10).join(sentiment_lines)}\n" + "Top 标签：\n" + f"{chr(10).join(label_lines)}\n" + "典型评论：\n" + f"{chr(10).join(typical_lines)}" + ) + + def parse_ai_comment_response(response_text: str, *, expected_comment_ids: set[str]) -> list[AIAnalysisResult]: try: raw = json.loads(response_text) @@ -118,7 +158,7 @@ class OpenAICompatibleAIClient: self.model = model self._http_client = http_client or httpx.Client(timeout=timeout_seconds) - def request(self, prompt: str) -> str: + def request(self, prompt: str, *, system_prompt: str = "请严格遵循用户指令输出。") -> str: endpoint = f"{self.base_url}/chat/completions" if self.base_url.endswith("/v1") else f"{self.base_url}/v1/chat/completions" response = self._http_client.post( endpoint, @@ -128,7 +168,7 @@ class OpenAICompatibleAIClient: "messages": [ { "role": "system", - "content": "只返回 JSON Array，不输出 Markdown 或解释性自然语言。", + "content": system_prompt, }, {"role": "user", "content": prompt}, ], diff --git a/app/services/report_service.py b/app/services/report_service.py index bf56848..9a278aa 100644 --- a/app/services/report_service.py +++ b/app/services/report_service.py @@ -128,7 +128,12 @@ def _safe_summary(summary_provider, metrics: dict, typical: dict, *, limit: int) if summary_provider is None: return DEFAULT_SUMMARY try: - summary = summary_provider(metrics, typical) + summary = summary_provider(metrics, typical, word_limit=limit) + except TypeError: + try: + summary = summary_provider(metrics, typical) + except Exception: + return DEFAULT_SUMMARY except Exception: return DEFAULT_SUMMARY return str(summary)[:limit] diff --git a/app/services/task_service.py b/app/services/task_service.py index 3d0c807..527c150 100644 --- a/app/services/task_service.py +++ b/app/services/task_service.py @@ -13,7 +13,7 @@ from app.platforms.base import PlatformAPIError, TikHubClient from app.platforms.douyin import DouyinPlatform from app.platforms.xiaohongshu import XiaohongshuPlatform from app.schemas import CreateTaskRequest -from app.services.ai_service import OpenAICompatibleAIClient, analyze_comments_with_retry, calculate_analysis_status +from app.services.ai_service import OpenAICompatibleAIClient, analyze_comments_with_retry, build_report_summary_prompt, calculate_analysis_status from app.services.report_service import generate_hotspot_report, generate_item_report @@ -52,16 +52,25 @@ def build_platform(platform: str): raise ValueError(f"Unsupported platform: {platform}") -def build_ai_requester(): +def build_ai_client() -> OpenAICompatibleAIClient | None: settings = get_settings() if settings.ai_base_url and settings.ai_api_key and settings.ai_model: - client = OpenAICompatibleAIClient( + return OpenAICompatibleAIClient( base_url=settings.ai_base_url, api_key=settings.ai_api_key, model=settings.ai_model, timeout_seconds=settings.ai_timeout_seconds, ) - return client.request + return None + + +def build_ai_requester(client: OpenAICompatibleAIClient | None = None): + client = client if client is not None else build_ai_client() + if client is not None: + return lambda prompt: client.request( + prompt, + system_prompt="只返回 JSON Array，不输出 Markdown 或解释性自然语言。", + ) def fallback_requester(_prompt: str) -> str: return "[]" @@ -69,6 +78,25 @@ def build_ai_requester(): return fallback_requester +def build_report_summary_provider(client: OpenAICompatibleAIClient | None): + if client is None: + return None + + def summarize_report(metrics: dict, typical: dict, *, word_limit: int = 200) -> str: + return client.request( + build_report_summary_prompt(metrics, typical, word_limit=word_limit), + system_prompt="只返回一段中文总结，不使用 Markdown。", + ) + + return summarize_report + + +def build_ai_dependencies(): + client = build_ai_client() + requester = build_ai_requester(client) + return requester, build_report_summary_provider(client) + + def create_task(session: Session, request: CreateTaskRequest, *, submit_background: bool = True) -> Task: task = Task( platform=request.platform, @@ -108,6 +136,8 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma return try: platform = build_platform(task.platform) + ai_requester, report_summary_provider = build_ai_dependencies() + try: hotspots = platform.fetch_hotspots(limit=task.hotspot_limit) except PlatformAPIError as exc: @@ -183,7 +213,7 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma item_comments = list(session.scalars(select(Comment).where(Comment.content_item_id == item.id))) ai_results = analyze_comments_with_retry( [{"comment_id": comment.id, "content": comment.content} for comment in item_comments], - requester=build_ai_requester(), + requester=ai_requester, max_retries=get_settings().ai_max_retries, ) results_by_comment_id = {result.comment_id: result for result in ai_results} @@ -201,7 +231,7 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma comment.ai_analysis_status = result.ai_analysis_status item.status = "success" task.successful_items_count += 1 - generate_item_report(session, item.id) + generate_item_report(session, item.id, summary_provider=report_summary_provider) except Exception as exc: item.status = "failed" item.error_stage = "crawl_comments" @@ -223,7 +253,7 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma total_count=total_comments, ) for hotspot in task.hotspots: - generate_hotspot_report(session, hotspot.id) + generate_hotspot_report(session, hotspot.id, summary_provider=report_summary_provider) task.status = "success" else: _mark_task_failed(task, task.error_stage or "crawl_items", task.error_type or "no_successful_items", task.error_message or "没有任何内容条目成功") diff --git a/docker-compose.yml b/docker-compose.yml index accce96..b646ac8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,3 +1,5 @@ +name: hot-comments-tool + services: app: build: . diff --git a/tests/integration/test_task_flow_xiaohongshu.py b/tests/integration/test_task_flow_xiaohongshu.py index a1ed6e6..239e0c8 100644 --- a/tests/integration/test_task_flow_xiaohongshu.py +++ b/tests/integration/test_task_flow_xiaohongshu.py @@ -5,6 +5,7 @@ from sqlalchemy.orm import Session from app.models import Comment, ContentItem, Hotspot, Report, Task from app.schemas import CreateTaskRequest +from app.services.report_service import DEFAULT_SUMMARY from app.services.task_service import create_task, run_task from tests.helpers import make_test_client @@ -28,7 +29,11 @@ class FakeXhsPlatform: return [CommentData(source_comment_id="c1", content="评论一", like_count=5, raw_data={"id": "c1"})] +ai_prompts: list[str] = [] + + def successful_ai_response(prompt: str) -> str: + ai_prompts.append(prompt) comments = json.loads(prompt[prompt.index("[") :]) comment_id = comments[0]["comment_id"] return json.dumps( @@ -38,9 +43,14 @@ def successful_ai_response(prompt: str) -> str: def test_xiaohongshu_task_flow_persists_hotspots_items_and_comments(monkeypatch): + ai_prompts.clear() with make_test_client() as (_client, engine): monkeypatch.setattr("app.services.task_service.build_platform", lambda _platform: FakeXhsPlatform()) - monkeypatch.setattr("app.services.task_service.build_ai_requester", lambda: successful_ai_response) + summary_provider = lambda metrics, typical, *, word_limit=200: ( + ai_prompts.append(f"只返回一段中文总结 word_limit={word_limit} sample={metrics['sample_count']}") + or "这是一段真实 AI 报告摘要。" + ) + monkeypatch.setattr("app.services.task_service.build_ai_dependencies", lambda: (successful_ai_response, summary_provider)) with Session(engine) as session: task = create_task( @@ -66,14 +76,21 @@ def test_xiaohongshu_task_flow_persists_hotspots_items_and_comments(monkeypatch) assert comment.sentiment == "positive" assert comment.labels == '["认可"]' assert comment.reason == "喜欢" - assert session.scalar(select(Report).where(Report.task_id == task_id, Report.report_type == "item")) is not None - assert session.scalar(select(Report).where(Report.task_id == task_id, Report.report_type == "hotspot")) is not None + item_report = session.scalar(select(Report).where(Report.task_id == task_id, Report.report_type == "item")) + hotspot_report = session.scalar(select(Report).where(Report.task_id == task_id, Report.report_type == "hotspot")) + assert item_report is not None + assert hotspot_report is not None + assert item_report.summary == "这是一段真实 AI 报告摘要。" + assert hotspot_report.summary == "这是一段真实 AI 报告摘要。" + assert DEFAULT_SUMMARY not in item_report.markdown_content + assert DEFAULT_SUMMARY not in hotspot_report.markdown_content + assert any("只返回一段中文总结" in prompt for prompt in ai_prompts) def test_xiaohongshu_task_flow_marks_comments_failed_when_ai_parse_fails(monkeypatch): with make_test_client() as (_client, engine): monkeypatch.setattr("app.services.task_service.build_platform", lambda _platform: FakeXhsPlatform()) - monkeypatch.setattr("app.services.task_service.build_ai_requester", lambda: (lambda _prompt: "not-json")) + monkeypatch.setattr("app.services.task_service.build_ai_dependencies", lambda: (lambda _prompt: "not-json", None)) monkeypatch.setattr("app.services.ai_service.time.sleep", lambda _seconds: None) with Session(engine) as session: @@ -107,7 +124,7 @@ def test_xiaohongshu_task_flow_keeps_item_success_when_ai_request_raises(monkeyp def failing_requester(_prompt): raise RuntimeError("ai unauthorized") - monkeypatch.setattr("app.services.task_service.build_ai_requester", lambda: failing_requester) + monkeypatch.setattr("app.services.task_service.build_ai_dependencies", lambda: (failing_requester, None)) with Session(engine) as session: task = create_task( diff --git a/tests/unit/test_ai_schema.py b/tests/unit/test_ai_schema.py index dd9c621..792af86 100644 --- a/tests/unit/test_ai_schema.py +++ b/tests/unit/test_ai_schema.py @@ -5,10 +5,12 @@ from app.services.ai_service import ( AIAnalysisResult, OpenAICompatibleAIClient, analyze_comments_with_retry, + build_report_summary_prompt, build_comment_prompt, calculate_analysis_status, parse_ai_comment_response, ) +from app.services.task_service import build_report_summary_provider def test_build_comment_prompt_contains_ids_and_truncates_content(): @@ -20,6 +22,35 @@ def test_build_comment_prompt_contains_ids_and_truncates_content(): assert "你" * 151 not in prompt +def test_build_report_summary_prompt_contains_stats_labels_and_truncated_typical_comments(): + prompt = build_report_summary_prompt( + { + "sample_count": 2, + "sentiment": { + "positive": {"count": 1, "pct": 50.0}, + "neutral": {"count": 0, "pct": 0.0}, + "negative": {"count": 1, "pct": 50.0}, + "unknown": {"count": 0, "pct": 0.0}, + }, + "top_labels": [{"name": "价格争议", "count": 2}], + }, + {"positive": [{"content": "好" * 200}], "negative": [{"content": "太贵"}]}, + word_limit=200, + ) + + assert "只返回一段中文总结" in prompt + assert "样本评论数量：2" in prompt + assert "positive: 1 (50.0%)" in prompt + assert "价格争议: 2" in prompt + assert "好" * 150 in prompt + assert "好" * 151 not in prompt + assert "JSON Array" not in prompt + + +def test_report_summary_provider_is_disabled_without_real_ai_client(): + assert build_report_summary_provider(client=None) is None + + def test_parse_ai_comment_response_validates_array_sentiment_labels_and_ids(): result = parse_ai_comment_response( '[{"comment_id":"c1","sentiment":"positive","labels":["质量好"],"reason":"认可"}]', @@ -94,6 +125,27 @@ def test_openai_compatible_client_posts_chat_completion_and_returns_message_cont assert captured["authorization"] == "Bearer test-ai-key" assert '"model":"test-model"' in captured["body"] assert "prompt text" in captured["body"] + assert "请严格遵循用户指令输出。" in captured["body"] + + +def test_openai_compatible_client_accepts_custom_system_prompt(): + captured = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = request.read().decode("utf-8") + return httpx.Response(200, json={"choices": [{"message": {"content": "报告摘要"}}]}) + + http_client = httpx.Client(transport=httpx.MockTransport(handler)) + client = OpenAICompatibleAIClient( + base_url="https://ai.example.com", + api_key="test-ai-key", + model="test-model", + http_client=http_client, + ) + + assert client.request("总结 prompt", system_prompt="只返回一段中文总结，不使用 Markdown。") == "报告摘要" + assert "只返回一段中文总结" in captured["body"] + assert "JSON Array" not in captured["body"] def test_openai_compatible_client_accepts_base_url_that_already_includes_v1(): diff --git a/tests/unit/test_deployment_config.py b/tests/unit/test_deployment_config.py index eff71ce..9764147 100644 --- a/tests/unit/test_deployment_config.py +++ b/tests/unit/test_deployment_config.py @@ -11,6 +11,12 @@ def test_docker_compose_reads_real_env_file_not_example(): assert "- .env.example" not in compose_content +def test_docker_compose_uses_fixed_project_name_for_stable_port_owner(): + compose_content = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") + + assert "name: hot-comments-tool" in compose_content + + def test_real_env_file_is_gitignored(): gitignore_lines = (ROOT / ".gitignore").read_text(encoding="utf-8").splitlines() diff --git a/tests/unit/test_douyin_mapping.py b/tests/unit/test_douyin_mapping.py index 893a776..d7ef227 100644 --- a/tests/unit/test_douyin_mapping.py +++ b/tests/unit/test_douyin_mapping.py @@ -118,3 +118,39 @@ def test_douyin_maps_null_comments_as_empty_list(): payload = {"data": {"comments": None}} assert platform.map_comments(payload, limit=10) == [] + + +class RecordingDouyinClient: + def __init__(self): + self.requests = [] + + def get(self, path, *, params=None): + self.requests.append((path, params)) + cursor = params["cursor"] + if cursor == 0: + return { + "data": { + "comments": [{"cid": f"c-{index}", "text": f"评论 {index}"} for index in range(20)], + "cursor": 20, + "has_more": 1, + } + } + return { + "data": { + "comments": [{"cid": f"c-{index}", "text": f"评论 {index}"} for index in range(20, 55)], + "cursor": 55, + "has_more": 0, + } + } + + +def test_douyin_fetch_comments_paginates_until_limit(): + client = RecordingDouyinClient() + platform = DouyinPlatform(client=client) + + comments = platform.fetch_comments("aweme-1", limit=50) + + assert len(comments) == 50 + assert comments[0].source_comment_id == "c-0" + assert comments[-1].source_comment_id == "c-49" + assert [request[1]["cursor"] for request in client.requests] == [0, 20] diff --git a/tests/unit/test_task_executor.py b/tests/unit/test_task_executor.py index 28f0b94..75e0521 100644 --- a/tests/unit/test_task_executor.py +++ b/tests/unit/test_task_executor.py @@ -1,6 +1,7 @@ from concurrent.futures import ThreadPoolExecutor from app.services import task_service +from app.config import Settings def test_task_executor_is_single_worker_thread_pool(): @@ -10,3 +11,12 @@ def test_task_executor_is_single_worker_thread_pool(): def test_run_task_entrypoint_is_synchronous_function(): assert task_service.inspect.iscoroutinefunction(task_service.run_task) is False + + +def test_ai_dependencies_do_not_generate_report_summary_provider_without_ai_config(monkeypatch): + monkeypatch.setattr("app.services.task_service.get_settings", lambda: Settings(_env_file=None)) + + requester, summary_provider = task_service.build_ai_dependencies() + + assert requester("prompt") == "[]" + assert summary_provider is None