215 lines
13 KiB
Python
215 lines
13 KiB
Python
"""Pure v0.3 result projection: goal fulfillment, artifacts and runtime differ.
|
|
|
|
``project`` does not read files, update tasks or contact a model/platform. The
|
|
caller may supply ``exports`` from workflows.list_exports to attest real file
|
|
availability; plain historic manifest rows cannot attest an existing file.
|
|
Person badges always contain status/label/detail/evidence; required missing
|
|
fields are objects with key/label/required. This is the frontend contract.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from urllib.parse import urlsplit
|
|
|
|
from backend.config import settings
|
|
from backend.prospecting.people import group_candidates, required_fields, target_progress
|
|
from backend.prospecting.discussion_scope import conditions as discussion_conditions
|
|
|
|
|
|
def _page_key(url: object) -> tuple[str, str] | None:
|
|
if not isinstance(url, str):
|
|
return None
|
|
try:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme != "https" or parsed.hostname not in {"www.xiaohongshu.com", "xiaohongshu.com"}:
|
|
return None
|
|
return ("xiaohongshu.com", parsed.path.rstrip("/"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _business_badge(task: dict, source_ids: list[str]) -> dict | None:
|
|
"""User-reported business feedback, separate from observed action results."""
|
|
labels = {"replied": "用户记录已回复", "need_confirmed": "用户记录已确认需求", "cooperated": "用户记录已合作",
|
|
"no_reply": "用户记录未回复", "not_relevant": "用户记录不匹配"}
|
|
events = [row for row in (task.get("artifacts") or {}).get("business_outcomes") or [] if isinstance(row, dict)
|
|
and row.get("source") == "user" and row.get("source_id") in source_ids and row.get("outcome") in labels]
|
|
if not events:
|
|
return None
|
|
event = events[-1]
|
|
return {"status": "user_" + event["outcome"], "label": labels[event["outcome"]],
|
|
"detail": "用户主动记录的业务结果，未经页面自动核验；获客归因未证实，不增加方法技术成功次数。",
|
|
"evidence": [{"source_id": event["source_id"]}], "source": "user", "at": event.get("at"),
|
|
"attribution": "unproven"}
|
|
|
|
|
|
def scope_met(task: dict) -> bool:
|
|
"""Whether the explicit acquisition scope is proved, independent of files.
|
|
|
|
A numeric goal is established by eligible source-bound people. Finite page
|
|
and route checks additionally need successful collection plus completed
|
|
per-source analysis. Old ``status=completed`` alone is never proof.
|
|
"""
|
|
inputs = task.get("inputs") or {}
|
|
artifacts = task.get("artifacts") or {}
|
|
progress = target_progress(task)
|
|
intent = inputs.get("intent", "discover")
|
|
if intent in {"follow", "message"}:
|
|
grants = [row for row in artifacts.get("outreach_grants") or [] if isinstance(row, dict)
|
|
and row.get("action") == intent and row.get("goal_version") == inputs.get("_goal_version", 1)
|
|
and row.get("authorized_by") == "user_batch_request" and row.get("status") in {"authorized", "completed"}]
|
|
if not grants or artifacts.get("outreach_pause"):
|
|
return False
|
|
grant = grants[-1]
|
|
targets = set(grant.get("source_ids") or [])
|
|
actions = [row for row in artifacts.get("action_ledger") or [] if isinstance(row, dict) and row.get("grant_id") == grant.get("id")]
|
|
resolved = {row.get("source_id") for row in actions if row.get("status") == "success"
|
|
or (intent == "follow" and row.get("status") == "skipped_already_followed")}
|
|
unresolved = any(row.get("status") in {"failed", "unknown", "prepared", "executing"} for row in actions)
|
|
return bool(targets) and targets <= resolved and not unresolved
|
|
if not progress["assessment_current"]:
|
|
return False
|
|
if any(not condition["met"] for condition in discussion_conditions(task)):
|
|
return False
|
|
if progress["target"] is not None:
|
|
return progress["reached"]
|
|
if intent == "export":
|
|
return False # Its validated file contract establishes completion.
|
|
history = artifacts.get("browser_history") or {}
|
|
if (history.get("is_successful") is False or artifacts.get("browser_verdict") in {"failed", "interrupted", "paused"}
|
|
or not (history.get("is_successful") is True or artifacts.get("browser_verdict") == "success")):
|
|
return False
|
|
sources = [row for row in artifacts.get("sources") or [] if isinstance(row, dict) and row.get("source_id") and row.get("content")]
|
|
decisions = {row.get("source_id") for row in artifacts.get("decisions") or [] if isinstance(row, dict)}
|
|
if not sources or artifacts.get("analysis_pending") or any(row["source_id"] not in decisions for row in sources):
|
|
return False
|
|
notes = {row.get("note_id") or _page_key(row.get("source_url")) or row["source_id"] for row in sources if row.get("source_type") == "note"}
|
|
if type(inputs.get("max_notes")) is int:
|
|
return len(notes) >= inputs["max_notes"]
|
|
requested = inputs.get("subject_urls") or []
|
|
if requested:
|
|
# Shared short links cannot be equated with a different destination
|
|
# without a recorded redirect; retain unknown instead of guessing.
|
|
observed = {_page_key(row.get("source_url")) for row in sources}
|
|
return all(_page_key(url) is not None and _page_key(url) in observed for url in requested)
|
|
if inputs.get("discovery_scope") == "route_pass" or inputs.get("intent") in {"inspect", "research"}:
|
|
coverage = artifacts.get("coverage") or {}
|
|
observed = set(coverage.get("search_keywords_observed") or [])
|
|
keywords = inputs.get("keywords") or []
|
|
return not coverage.get("search_keywords_not_verified") and bool(keywords) and all(word in observed for word in keywords)
|
|
return False
|
|
|
|
|
|
def project(task: dict, *, exports: list[dict] | None = None) -> dict:
|
|
"""Return a view of evidence without rewriting legacy runtime outcomes."""
|
|
from backend.prospecting.workflows import result_version
|
|
from backend.prospecting.preflight import delivery_projection
|
|
|
|
preflight = delivery_projection(task) if settings.reply_relation_preflight_enabled else None
|
|
if preflight is not None:
|
|
return preflight
|
|
|
|
artifacts = task.get("artifacts") or {}
|
|
inputs = task.get("inputs") or {}
|
|
grouped = group_candidates(task)
|
|
progress = target_progress(task)
|
|
fields = required_fields(task)
|
|
people = []
|
|
for group in grouped["groups"]:
|
|
person = {key: group[key] for key in ("person_key", "eligible", "assessment_current", "group", "badges", "missing")}
|
|
business = _business_badge(task, group["source_ids"])
|
|
if business:
|
|
person["badges"] = {**person["badges"], "business": business}
|
|
people.append(person)
|
|
source_rows = [row for row in artifacts.get("sources") or [] if isinstance(row, dict) and row.get("content")]
|
|
discussion = discussion_conditions(task)
|
|
decisions = [row for row in artifacts.get("decisions") or [] if isinstance(row, dict)]
|
|
target = progress["target"]
|
|
current = progress["current"]
|
|
fulfilled_scope = scope_met(task)
|
|
file_required = "csv" in fields
|
|
version = result_version(task)
|
|
export_rows = exports if exports is not None else artifacts.get("exports") or []
|
|
current_files = [row for row in export_rows if row.get("available") is True
|
|
and row.get("is_current") is True and row.get("result_version") == version
|
|
and row.get("goal_version") == inputs.get("_goal_version", 1)]
|
|
file_available = bool(current_files)
|
|
known = grouped["counts"]["candidates"]
|
|
counts = {"eligible": current, "known_identity": known,
|
|
"supplemental": sum(person["group"] == "supplemental" for person in people),
|
|
"unresolved": sum(person["group"] == "unresolved" for person in people),
|
|
"signals": grouped["counts"]["signals"], "verified_accounts": grouped["counts"]["verified_accounts"]}
|
|
conditions = []
|
|
if target is not None:
|
|
conditions.append({"key": "candidate_count", "label": "符合本次条件的去重人数", "required": True,
|
|
"met": fulfilled_scope, "current": current, "target": target})
|
|
elif inputs.get("intent") not in {"export", "follow", "message"}:
|
|
conditions.append({"key": "scope", "label": "本次明确检查范围", "required": True,
|
|
"met": fulfilled_scope, "current": None, "target": None})
|
|
if people:
|
|
conditions.append({"key": "identity", "label": "来源身份可区分", "required": True,
|
|
"met": counts["unresolved"] == 0, "current": known, "target": len(people)})
|
|
conditions.extend(discussion)
|
|
for key, label, badge, status in (
|
|
("xiaohongshu_id", "已取得且正确归属的小红书号", "account", "obtained"),
|
|
("explicit_demand", "本人明确需求原话", "demand", "explicit"),
|
|
("region", "本人原话支持指定地区", "region", "matched"),
|
|
):
|
|
obtained = sum(person["badges"][badge]["status"] == status and person["group"] != "unresolved" for person in people) if progress["assessment_current"] else 0
|
|
finite_field_met = fulfilled_scope and all(person["group"] != "unresolved" and person["badges"][badge]["status"] == status for person in people)
|
|
conditions.append({"key": key, "label": label, "required": key in fields,
|
|
"met": obtained >= target if target is not None else finite_field_met if key in fields else obtained > 0,
|
|
"current": obtained, "target": target if key in fields else None})
|
|
conditions.append({"key": "csv", "label": "当前结果版本的CSV文件", "required": file_required,
|
|
"met": file_available, "current": len(current_files), "target": 1 if file_required else None})
|
|
list_available = bool(people)
|
|
# Missing fields or unclear judgments do not establish a negative business
|
|
# conclusion. Only actual exclusions across the read scope can prove an
|
|
# empty result; supplemental and unresolved groups remain available.
|
|
by_source = {row.get("source_id"): row for row in decisions}
|
|
rejected_scope = bool(source_rows) and all(by_source.get(row.get("source_id"), {}).get("decision") == "exclude" for row in source_rows)
|
|
conclusion_available = fulfilled_scope and not people and rejected_scope
|
|
requirements_met = all(condition["met"] for condition in conditions if condition["required"] and condition["key"] != "scope")
|
|
if inputs.get("intent", "discover") in {"discover", "research", "inspect"} and not people and not conclusion_available:
|
|
requirements_met = False
|
|
complete = fulfilled_scope and requirements_met
|
|
if inputs.get("intent") == "export":
|
|
complete = file_available
|
|
requirements_met = file_available
|
|
task_status = task.get("status") or "unknown"
|
|
running = task_status == "running"
|
|
stale = not progress["assessment_current"]
|
|
if inputs.get("intent") == "export" and file_available:
|
|
status, headline = "fulfilled", "文件已生成"
|
|
elif inputs.get("intent") in {"follow", "message"} and complete:
|
|
status, headline = "fulfilled", "跟进已完成"
|
|
elif stale and list_available:
|
|
status, headline = "unknown", "已有名单保留 本次标准尚未重新分析"
|
|
elif running:
|
|
status = "in_progress"
|
|
headline = (f"已整理{current}位 正在继续" if current else f"已保存{len(people)}组来源 正在继续") if people else "正在读取与整理资料"
|
|
elif complete and discussion:
|
|
status, headline = "fulfilled", "本次讨论检查已完成"
|
|
elif complete and current:
|
|
status, headline = "fulfilled", f"名单已交付 共{current}位"
|
|
elif complete and conclusion_available:
|
|
status, headline = "no_matches", "本次检查已完成 未发现符合条件的候选"
|
|
elif list_available or (discussion and source_rows):
|
|
status = "partial"
|
|
headline = ((f"阶段成果已交付 共{current}位" + (f" 目标{target}位" if target is not None else "")) if current
|
|
else f"阶段资料已交付 共{len(people)}组来源") if list_available else f"阶段资料已交付 共{len(source_rows)}条来源"
|
|
elif not source_rows:
|
|
status, headline = "no_evidence", "本阶段未取得可判断的资料"
|
|
else:
|
|
status, headline = "unknown", "已有资料保留 本次范围尚未完成"
|
|
history = artifacts.get("browser_history") or {}
|
|
return {"version": 1, "status": status, "headline": headline,
|
|
"runtime": {"status": task_status, "step": task.get("current_step"),
|
|
"stop_reason": artifacts.get("stop_reason"), "stop_detail": artifacts.get("stop_detail") or task.get("pause_reason") or "",
|
|
"raw_browser_success": history.get("is_successful"), "browser_verdict": artifacts.get("browser_verdict")},
|
|
"fulfillment": {"scope_met": fulfilled_scope, "requirements_met": requirements_met, "complete": complete,
|
|
"assessment_current": progress["assessment_current"], "target": target, "current": current,
|
|
"remaining": progress["remaining"]},
|
|
"artifacts": {"list_available": list_available, "file_required": file_required, "file_available": file_available,
|
|
"delivered": list_available or conclusion_available or file_available or bool(discussion and source_rows), "export_count": len(current_files)},
|
|
"counts": counts, "conditions": conditions, "people": people}
|