386 lines
21 KiB
Python
386 lines
21 KiB
Python
"""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"]
|