"""v0.3 delivery acceptance: pure fixtures, no real DB/model/platform calls.""" from copy import deepcopy import csv from pathlib import Path import pytest from backend.prospecting import delivery, dialogue, people, workflows from backend.prospecting.analysis import validate_analysis from backend.prospecting.contracts import ProspectingBrief def task(**inputs): return {"id": "delivery-v03-fixture", "status": "completed", "current_step": "results", "inputs": {**ProspectingBrief(target_candidates=2).model_dump(), "_goal_version": 1, **inputs}, "artifacts": {"sources": [], "leads": [], "decisions": [], "profiles": [], "analysis_goal_version": 1}} def add(row, number, *, account=False, identity=True, quote="我最近准备装修", **analysis): sid = "s" + str(number) url = "https://www.xiaohongshu.com/user/profile/" + f"{number:024x}" source = {"source_id": sid, "source_type": "comment", "author": "同名", "content": quote, "source_url": "https://www.xiaohongshu.com/explore/n", "observed_at": "2026-10-29T10:00:00+08:00"} if identity: source["author_profile_url"] = url lead = {"source_id": sid, "id": sid, "decision": "keep", "quote": quote, "reason": "本人相关表达", **analysis} row["artifacts"]["sources"].append(source) row["artifacts"]["leads"].append(lead) row["artifacts"]["decisions"].append(deepcopy(lead)) if account: row["artifacts"]["profiles"].append({"source_id": sid, "profile_url": url, "xiaohongshu_id": f"public_{number}", "observed_at": "2026-10-29T10:00:00+08:00", "evidence": "matched_source_author_link_and_visible_profile_header"}) return source, lead def test_optional_unknown_fields_do_not_prevent_twenty_person_delivery(): row = task(target_candidates=20) for i in range(20): add(row, i + 1, seeking_service="yes" if i < 12 else "unknown") before = deepcopy(row) view = delivery.project(row) assert view["headline"] == "名单已交付 共20位" assert view["fulfillment"]["complete"] is True assert view["counts"]["eligible"] == 20 assert sum(person["badges"]["demand"]["status"] == "explicit" for person in view["people"]) == 12 assert all(person["badges"]["account"]["status"] == "unknown" for person in view["people"]) assert row == before def test_explicit_demand_requirement_keeps_related_signals_supplemental(): row = task(target_candidates=20, required_result_fields=["explicit_demand"]) for i in range(20): add(row, i + 1, seeking_service="yes" if i < 12 else "unknown") progress = people.target_progress(row) view = delivery.project(row) assert (progress["current"], progress["remaining"], progress["known_identity"]) == (12, 8, 20) assert view["status"] == "partial" and view["counts"]["supplemental"] == 8 assert not delivery.scope_met(row) def test_only_owned_visible_accounts_count_for_explicit_account_goal(): row = task(required_result_fields=["xiaohongshu_id"]) add(row, 1, account=True) source, _ = add(row, 2) row["artifacts"]["profiles"].append({"source_id": source["source_id"], "profile_id": "internal-id", "profile_url": source["author_profile_url"], "xiaohongshu_id": "guessed-public"}) assert people.target_progress(row)["current"] == 1 row["artifacts"]["profiles"][-1].update(observed_at="now", evidence="matched_source_author_link_and_visible_profile_header", profile_url=row["artifacts"]["sources"][0]["author_profile_url"]) assert people.target_progress(row)["current"] == 1 def test_region_requires_same_source_quote_never_ip_or_parent_inheritance(): row = task(region="杭州", target_candidates=1) _, lead = add(row, 1, seeking_service="yes", region_match="yes", region_quote="杭州找装修", region_scope="杭州") row["artifacts"]["sources"][0]["ip_location"] = "杭州" assert people.target_progress(row)["current"] == 0 lead.update(region_quote="我最近准备装修") # Semantic region output is accepted only when bound to this target and # actually supported by an own-source region quote (analysis enforces it). lead.update(region_quote="") assert delivery.project(row)["people"][0]["badges"]["region"]["status"] == "unknown" row["artifacts"]["sources"][0]["content"] = "杭州找装修" lead.update(quote="杭州找装修", region_quote="杭州找装修") assert people.target_progress(row)["current"] == 1 @pytest.mark.parametrize("region_quote", ["父评论杭州找装修", "IP属地：杭州", "杭州", ""]) def test_analysis_rejects_inherited_ip_or_missing_region_evidence(region_quote): source = {"source_id": "reply", "source_type": "reply", "content": "我也想问问 IP属地：杭州"} payload = {"results": [{"source_id": "reply", "decision": "keep", "quote": "我也想问问", "reason": "相关信号", "region_match": "yes", "region_quote": region_quote}]} with pytest.raises(ValueError): validate_analysis(payload, [source], ProspectingBrief(region="杭州")) def test_valid_analysis_records_region_scope_with_own_source_quote(): source = {"source_id": "s", "content": "杭州的房子要装修，求设计师"} decisions, _ = validate_analysis({"results": [{"source_id": "s", "decision": "keep", "quote": source["content"], "reason": "本人需求", "seeking_service": "yes", "region_match": "yes", "region_quote": "杭州的房子"}]}, [source], ProspectingBrief(region="杭州")) assert decisions[0]["region_scope"] == "杭州" def test_generic_own_quote_cannot_be_used_as_location_proof(): source = {"source_id": "s", "content": "我准备装修，想找设计师"} payload = {"results": [{"source_id": "s", "decision": "keep", "quote": source["content"], "reason": "本人需求", "seeking_service": "yes", "region_match": "yes", "region_quote": "我准备装修"}]} with pytest.raises(ValueError, match="地点原话"): validate_analysis(payload, [source], ProspectingBrief(region="杭州")) row = task(region="杭州", target_candidates=1) add(row, 1, region_match="yes", region_quote="我最近准备装修", region_scope="杭州") assert people.target_progress(row)["current"] == 0 def test_identity_unknown_and_duplicate_person_do_not_pad_goal(): row = task() add(row, 1) add(row, 2, identity=False) duplicate = deepcopy(row["artifacts"]["leads"][0]) duplicate.update(source_id="another", id="another") source = {**row["artifacts"]["sources"][0], "source_id": "another"} row["artifacts"]["sources"].append(source) row["artifacts"]["leads"].append(duplicate) view = delivery.project(row) assert view["counts"]["eligible"] == 1 and view["counts"]["unresolved"] == 1 assert people.group_candidates(row)["counts"]["signals"] == 3 def current_export(row): return {"id": "fixture", "available": True, "is_current": True, "goal_version": 1, "result_version": workflows.result_version(row)} def test_interrupted_eight_of_twenty_with_file_is_partial_and_preserves_raw_failure(): row = task(target_candidates=20, required_result_fields=["csv"]) for i in range(8): add(row, i + 1) row.update(status="failed", current_step="interrupted") row["artifacts"].update(browser_history={"is_successful": False}, stop_reason="execution_error", stop_detail="实际执行异常") before = deepcopy(row) view = delivery.project(row, exports=[current_export(row)]) assert view["headline"] == "阶段成果已交付 共8位 目标20位" assert view["artifacts"]["file_available"] and not view["fulfillment"]["complete"] assert view["runtime"]["raw_browser_success"] is False and view["runtime"]["status"] == "failed" assert row == before def test_file_requirement_does_not_disappear_at_candidate_target(): row = task(target_candidates=1, required_result_fields=["csv"]) add(row, 1) assert delivery.scope_met(row) assert not delivery.project(row)["fulfillment"]["complete"] assert delivery.project(row, exports=[current_export(row)])["fulfillment"]["complete"] def test_export_only_file_can_complete_while_acquisition_and_legacy_analysis_remain_separate(): row = task(intent="export", target_candidates=None, _goal_version=2) add(row, 1) exported = {**current_export(row), "goal_version": 2} view = delivery.project(row, exports=[exported]) assert view["headline"] == "文件已生成" and view["fulfillment"]["complete"] assert not view["fulfillment"]["assessment_current"] assert not view["fulfillment"]["scope_met"] and row["artifacts"]["analysis_goal_version"] == 1 def test_stale_file_or_stale_analysis_cannot_fulfill_new_goal(): row = task(target_candidates=1, required_result_fields=["csv"]) add(row, 1) exported = current_export(row) row["inputs"]["_goal_version"] = 2 view = delivery.project(row, exports=[exported]) assert not view["fulfillment"]["complete"] and not view["fulfillment"]["assessment_current"] assert not view["artifacts"]["file_available"] and view["counts"]["eligible"] == 0 assert view["artifacts"]["list_available"] and len(view["people"]) == 1 def test_running_task_with_partial_results_has_no_early_completed_title(): row = task() row["status"] = "running" add(row, 1) assert delivery.project(row)["headline"] == "已整理1位 正在继续" @pytest.mark.parametrize("duplicate_evidence", [False, True]) def test_legacy_zero_eligible_retains_supplemental_and_unresolved_groups_without_counting_them_as_people(duplicate_evidence): row = task(target_candidates=None, region="杭州") row["artifacts"].pop("analysis_goal_version") row["artifacts"].update(browser_history={"is_successful": False}, browser_verdict="failed", stop_reason="agent_reported_failure") add(row, 1) for number in range(2, 6): add(row, number, identity=False) if duplicate_evidence: source = {**row["artifacts"]["sources"][0], "source_id": "same-person-second-source"} lead = {**row["artifacts"]["leads"][0], "source_id": source["source_id"], "id": source["source_id"]} row["artifacts"]["sources"].append(source) row["artifacts"]["leads"].append(lead) row["artifacts"]["decisions"].append(deepcopy(lead)) before = deepcopy(row) view = delivery.project(row, exports=[current_export(row)]) assert view["headline"] == "阶段资料已交付 共5组来源" and view["status"] == "partial" assert view["counts"]["eligible"] == 0 and view["counts"]["supplemental"] == 1 and view["counts"]["unresolved"] == 4 assert view["counts"]["signals"] == (6 if duplicate_evidence else 5) assert view["artifacts"]["file_available"] is True and view["artifacts"]["list_available"] is True assert view["fulfillment"]["complete"] is False and view["runtime"]["raw_browser_success"] is False assert row == before row["status"] = "running" running = delivery.project(row) assert running["status"] == "in_progress" and running["headline"] == "已保存5组来源 正在继续" assert running["fulfillment"]["complete"] is False row["inputs"]["_goal_version"] = 2 row["artifacts"]["analysis_goal_version"] = 1 stale = delivery.project(row) assert stale["status"] == "unknown" and stale["headline"] == "已有名单保留 本次标准尚未重新分析" def test_zero_group_materials_are_distinct_from_no_evidence_and_one_eligible_result(): row = task(target_candidates=2, required_result_fields=["explicit_demand"]) add(row, 1) assert delivery.project(row)["headline"] == "阶段资料已交付 共1组来源" add(row, 2, seeking_service="yes") assert delivery.project(row)["headline"] == "阶段成果已交付 共1位 目标2位" empty = delivery.project(task()) assert empty["status"] == "no_evidence" and empty["headline"] == "本阶段未取得可判断的资料" assert not empty["artifacts"]["delivered"] @pytest.mark.parametrize("identity", [True, False]) def test_finished_finite_read_with_unknown_required_fields_delivers_materials_not_false_no_matches(identity): url = "https://www.xiaohongshu.com/explore/n" row = task(intent="inspect", target_candidates=None, subject_urls=[url], region="杭州") add(row, 1, identity=identity) row["artifacts"].update(browser_history={"is_successful": True}, browser_verdict="success") view = delivery.project(row) assert view["fulfillment"]["scope_met"] is True and view["fulfillment"]["complete"] is False assert view["status"] == "partial" and view["headline"] == "阶段资料已交付 共1组来源" region = next(item for item in view["conditions"] if item["key"] == "region") assert region["required"] and not region["met"] assert view["counts"]["eligible"] == 0 and view["people"][0]["missing"] assert view["counts"]["supplemental"] == int(identity) and view["counts"]["unresolved"] == int(not identity) def test_finite_read_unknown_identity_cannot_be_a_negative_result_even_without_extra_required_fields(): url = "https://www.xiaohongshu.com/explore/n" row = task(intent="inspect", target_candidates=None, subject_urls=[url]) add(row, 1, identity=False) row["artifacts"].update(browser_history={"is_successful": True}, browser_verdict="success") view = delivery.project(row) assert view["status"] == "partial" and not view["fulfillment"]["complete"] identity = next(item for item in view["conditions"] if item["key"] == "identity") assert identity["required"] and identity["met"] is False def test_finite_read_unclear_decision_stays_unknown_until_actual_exclusion_is_available(): url = "https://www.xiaohongshu.com/explore/n" row = task(intent="inspect", target_candidates=None, subject_urls=[url]) row["artifacts"].update(browser_history={"is_successful": True}, browser_verdict="success", sources=[{"source_id": "n", "source_type": "note", "source_url": url, "content": "信息不完整"}], decisions=[{"source_id": "n", "decision": "unclear"}]) unknown = delivery.project(row) assert unknown["status"] == "unknown" and not unknown["fulfillment"]["complete"] row["artifacts"]["decisions"][0]["decision"] = "exclude" excluded = delivery.project(row) assert excluded["status"] == "no_matches" and excluded["fulfillment"]["complete"] def test_finite_read_with_no_matches_differs_from_unread_page(): url = "https://www.xiaohongshu.com/explore/n" row = task(intent="inspect", target_candidates=None, subject_urls=[url], region="杭州") row["artifacts"].update(browser_history={"is_successful": True}, browser_verdict="success") assert delivery.project(row)["status"] == "no_evidence" row["artifacts"]["sources"] = [{"source_id": "n", "source_type": "note", "content": "已经装修完成的分享", "source_url": url}] row["artifacts"]["decisions"] = [{"source_id": "n", "decision": "exclude"}] assert delivery.project(row)["status"] == "no_matches" row["artifacts"]["browser_history"]["is_successful"] = False assert delivery.project(row)["status"] == "unknown" and not delivery.scope_met(row) def test_legacy_completed_flag_is_not_proof_of_inspected_scope(): row = task(target_candidates=None, discovery_scope="route_pass", keywords=["装修"]) add(row, 1) assert delivery.project(row)["status"] == "partial" assert not delivery.scope_met(row) @pytest.mark.parametrize("message,fields", [ ("整理20个可见小红书号", {"xiaohongshu_id"}), ("只接受本人有明确需求的人，给我导出CSV", {"explicit_demand", "csv"}), ("找装修候选，账号可选，不一定有明确需求", set()), ("例如：整理小红书号\n> 只接受明确需求\n找候选", set()), ('用户原话“必须有小红书号”只是示例，找候选', set()), ("我是做装修的，找客户", set()), ]) def test_required_fields_are_grounded_in_actual_user_instructions(message, fields): proposed = ProspectingBrief(required_result_fields=["xiaohongshu_id", "explicit_demand", "region", "csv"], region="杭州") grounded = dialogue._ground_scope(proposed, ProspectingBrief(), message, []) assert set(grounded.required_result_fields) == fields assert grounded.region is None if "20个" in message: assert grounded.target_candidates == 20 def test_numeric_continuation_inherits_requirements_and_user_can_remove_them(): current = ProspectingBrief(target_candidates=20, required_result_fields=["xiaohongshu_id", "explicit_demand", "csv", "region"], region="杭州") numeric = dialogue._ground_scope(ProspectingBrief(), current, "5", []) assert set(numeric.required_result_fields) == set(current.required_result_fields) and numeric.region == "杭州" revised = dialogue._ground_scope(ProspectingBrief(), current, "不需要小红书号，不必有明确需求，不用导出，不限制地区", []) assert revised.required_result_fields == [] and revised.region is None @pytest.mark.parametrize("message,region", [ ("找杭州装修客户", "杭州"), ("地区改为杭州", "杭州"), ("我是杭州商家，找装修客户", None), ("最好杭州，找装修客户", None), ("参考示例：\n> 找杭州装修客户\n给我找客户", None), ]) def test_region_filter_requires_raw_user_goal_scope_not_merchant_location_or_examples(message, region): brief = dialogue._ground_scope(ProspectingBrief(region="杭州"), ProspectingBrief(), message, []) assert brief.region == region assert ("region" in brief.required_result_fields) is bool(region) def test_follow_and_send_have_separate_real_ledger_states(): row = task(target_candidates=1) add(row, 1) assert delivery.project(row)["people"][0]["badges"]["outreach"]["status"] == "not_executed" row["artifacts"]["action_ledger"] = [{"source_id": "s1", "action": "follow", "status": "success"}] assert delivery.project(row)["people"][0]["badges"]["outreach"]["status"] == "followed" row["artifacts"]["action_ledger"].append({"source_id": "s1", "action": "message", "status": "success"}) assert delivery.project(row)["people"][0]["badges"]["outreach"]["label"] == "已发送" row["artifacts"]["action_ledger"].append({"source_id": "s1", "action": "message", "status": "unknown"}) assert delivery.project(row)["people"][0]["badges"]["outreach"]["label"] == "结果未知" @pytest.mark.parametrize("outcome,label", [ ("replied", "用户记录已回复"), ("need_confirmed", "用户记录已确认需求"), ("cooperated", "用户记录已合作"), ("no_reply", "用户记录未回复"), ("not_relevant", "用户记录不匹配"), ]) def test_user_business_feedback_is_separate_from_action_evidence_and_qualification(outcome, label): row = task(target_candidates=1, required_result_fields=["explicit_demand"]) add(row, 1) row["artifacts"]["action_ledger"] = [{"source_id": "s1", "action": "message", "status": "unknown"}] row["artifacts"]["business_outcomes"] = [{"source_id": "s1", "outcome": outcome, "source": "user", "at": "2026-10-29T10:00:00+08:00", "attribution": "unproven"}] before = deepcopy(row) view = delivery.project(row) person = view["people"][0] assert person["badges"]["business"]["label"] == label assert person["badges"]["business"]["source"] == "user" assert person["badges"]["business"]["attribution"] == "unproven" assert person["badges"]["outreach"]["label"] == "结果未知" assert person["badges"]["demand"]["status"] == "related" assert not person["eligible"] and view["counts"]["eligible"] == 0 assert row == before def test_feedback_for_another_source_or_model_report_does_not_change_person_badges(): row = task(target_candidates=1) add(row, 1) row["artifacts"]["business_outcomes"] = [ {"source_id": "other", "outcome": "replied", "source": "user"}, {"source_id": "s1", "outcome": "cooperated", "source": "model"}, ] assert "business" not in delivery.project(row)["people"][0]["badges"] row["artifacts"]["business_outcomes"].extend([ {"source_id": "s1", "outcome": "replied", "source": "user"}, {"source_id": "s1", "outcome": "not_relevant", "source": "user"}, ]) assert delivery.project(row)["people"][0]["badges"]["business"]["status"] == "user_not_relevant" def test_csv_preserves_old_columns_and_matches_delivery_qualification(tmp_path, monkeypatch): row = task(required_result_fields=["xiaohongshu_id"]) add(row, 1, account=True) add(row, 2) monkeypatch.setattr(workflows, "task_doc_dir", lambda task_id: tmp_path) monkeypatch.setattr(workflows.db, "update_task", lambda task_id, **updates: row.update(updates)) entry = workflows.export_leads(row) with Path(entry["path"]).open(encoding="utf-8-sig", newline="") as stream: reader = csv.DictReader(stream) assert reader.fieldnames[:9] == ["昵称", "小红书号", "主页链接", "需求原话", "来源链接", "观察时间", "判断依据", "跟进状态", "用户纠正"] rows = list(reader) assert entry["count"] == 2 and entry["candidate_count"] == delivery.project(row)["counts"]["eligible"] == 1 assert [item["交付分组"] for item in rows] == ["符合本次条件", "补充来源"] assert all(item["跟进状态"] == "未执行" for item in rows) assert delivery.project(row, exports=workflows.list_exports(row))["artifacts"]["file_available"]