294 lines
9.9 KiB
Python
294 lines
9.9 KiB
Python
from copy import deepcopy
|
|
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from rest_framework.exceptions import ValidationError
|
|
|
|
from .models import StoryChoice, StoryMark, StoryRun, StoryVersion
|
|
|
|
|
|
def score_mathbti(definition, answer_indexes):
|
|
questions = definition.get("questions", [])
|
|
if len(answer_indexes) != len(questions):
|
|
raise ValidationError({"answers": "答案数量与题目数量不一致"})
|
|
|
|
axis_scores = {axis["id"]: 0 for axis in definition.get("axes", [])}
|
|
for question, raw_index in zip(questions, answer_indexes):
|
|
try:
|
|
option_index = int(raw_index)
|
|
options = question["options"]
|
|
if option_index < 0 or option_index >= len(options):
|
|
raise IndexError
|
|
option = options[option_index]
|
|
except (IndexError, KeyError, TypeError, ValueError) as exc:
|
|
raise ValidationError({"answers": f"题目 {question.get('id')} 的答案无效"}) from exc
|
|
axis_scores[question["axis"]] += int(option["score"])
|
|
|
|
cutoff = int(definition.get("scoring", {}).get("cutoff", 4))
|
|
axis_order = definition.get("scoring", {}).get("axes", list(axis_scores))
|
|
identity_code = "".join("1" if axis_scores[axis] > cutoff else "0" for axis in axis_order)
|
|
return identity_code, axis_scores
|
|
|
|
|
|
def validate_story_content(content):
|
|
nodes = content.get("nodes")
|
|
start = content.get("start_node")
|
|
errors = []
|
|
if not isinstance(nodes, dict) or not nodes:
|
|
return ["nodes 必须是非空对象"]
|
|
if start not in nodes:
|
|
errors.append("start_node 不存在")
|
|
|
|
referenced = set()
|
|
mark_codes = {
|
|
item.get("code")
|
|
for item in content.get("mark_definitions", [])
|
|
if isinstance(item, dict)
|
|
}
|
|
for node_id, node in nodes.items():
|
|
if not isinstance(node.get("choices", []), list):
|
|
errors.append(f"{node_id}.choices 必须是数组")
|
|
continue
|
|
for choice in node.get("choices", []):
|
|
target = choice.get("next")
|
|
if not target:
|
|
errors.append(f"{node_id} 存在缺少 next 的选项")
|
|
elif target not in nodes:
|
|
errors.append(f"{node_id} 引用了不存在的节点 {target}")
|
|
else:
|
|
referenced.add(target)
|
|
required = choice.get("requires_marks", [])
|
|
if not isinstance(required, list) or any(
|
|
code not in mark_codes for code in required
|
|
):
|
|
errors.append(f"{node_id} 的印记条件无效")
|
|
effects = choice.get("effects", {})
|
|
if content.get("system_version") == "marks-v1" and any(
|
|
key not in {"marks", "easter_eggs"} for key in effects
|
|
):
|
|
errors.append(f"{node_id} 仍包含旧数值效果")
|
|
|
|
if start in nodes:
|
|
reachable = set()
|
|
pending = [start]
|
|
while pending:
|
|
node_id = pending.pop()
|
|
if node_id in reachable:
|
|
continue
|
|
reachable.add(node_id)
|
|
pending.extend(
|
|
choice.get("next")
|
|
for choice in nodes[node_id].get("choices", [])
|
|
if choice.get("next") in nodes
|
|
)
|
|
ending_rules = content.get("ending_rules", {})
|
|
if node_id == ending_rules.get("trigger_node"):
|
|
pending.extend(ending_rules.get("domains", {}).values())
|
|
unreachable = sorted(set(nodes) - reachable)
|
|
if unreachable:
|
|
errors.append(f"存在不可达节点: {', '.join(unreachable[:10])}")
|
|
return errors
|
|
|
|
|
|
def _merge_effects(state, effects):
|
|
result = deepcopy(state)
|
|
for key, value in effects.items():
|
|
if isinstance(value, dict):
|
|
result[key] = _merge_effects(result.get(key, {}), value)
|
|
elif isinstance(value, list):
|
|
current = result.get(key, [])
|
|
result[key] = list(dict.fromkeys([*current, *value]))
|
|
elif isinstance(value, (int, float)):
|
|
result[key] = result.get(key, 0) + value
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def _mark_lookup(content):
|
|
return {
|
|
item["code"]: item
|
|
for item in content.get("mark_definitions", [])
|
|
if isinstance(item, dict) and item.get("code")
|
|
}
|
|
|
|
|
|
def _refresh_mark_counts(state, content):
|
|
result = deepcopy(state)
|
|
lookup = _mark_lookup(content)
|
|
counts = {
|
|
domain: 0 for domain in content.get("domain_labels", {})
|
|
}
|
|
for code in result.get("marks", []):
|
|
mark = lookup.get(code)
|
|
if mark:
|
|
counts[mark["domain"]] = counts.get(mark["domain"], 0) + 1
|
|
result["mark_counts"] = counts
|
|
return result
|
|
|
|
|
|
def _available_choices(run, node):
|
|
marks = set(run.state.get("marks", []))
|
|
return [
|
|
choice
|
|
for choice in node.get("choices", [])
|
|
if set(choice.get("requires_marks", [])).issubset(marks)
|
|
]
|
|
|
|
|
|
def _resolve_ending(content, state, node_id):
|
|
rules = content.get("ending_rules", {})
|
|
if node_id != rules.get("trigger_node"):
|
|
return node_id
|
|
counts = state.get("mark_counts", {})
|
|
order = rules.get("tie_order", [])
|
|
domain = max(order, key=lambda item: counts.get(item, 0))
|
|
return rules.get("domains", {}).get(domain, node_id)
|
|
|
|
|
|
def _node_payload(run):
|
|
content = run.story_version.content
|
|
node = content["nodes"][run.current_node]
|
|
available_choices = _available_choices(run, node)
|
|
choices = [
|
|
{
|
|
"index": index,
|
|
"text": choice.get("text", ""),
|
|
"special": bool(choice.get("requires_marks")),
|
|
}
|
|
for index, choice in enumerate(available_choices)
|
|
]
|
|
chapter_code = node.get("chapter") or run.current_node.split("_", 1)[0]
|
|
chapter = content.get("chapters", {}).get(chapter_code, {})
|
|
collection = list(
|
|
StoryMark.objects.filter(
|
|
user=run.user,
|
|
story=run.story_version.story,
|
|
).values("code", "name", "domain", "acquired_at")
|
|
)
|
|
return {
|
|
"run_id": run.id,
|
|
"story": run.story_version.story.title,
|
|
"version": run.story_version.version,
|
|
"status": run.status,
|
|
"current_node": run.current_node,
|
|
"state": run.state,
|
|
"chapter": chapter,
|
|
"mark_definitions": content.get("mark_definitions", []),
|
|
"domain_labels": content.get("domain_labels", {}),
|
|
"collection": collection,
|
|
"node": {
|
|
"scene": node.get("scene", ""),
|
|
"character": node.get("character", ""),
|
|
"sprite": node.get("sprite", ""),
|
|
"choices": choices,
|
|
},
|
|
}
|
|
|
|
|
|
@transaction.atomic
|
|
def start_story(user, story):
|
|
version = (
|
|
StoryVersion.objects.select_related("story")
|
|
.filter(story=story, is_published=True)
|
|
.order_by("-version")
|
|
.first()
|
|
)
|
|
if version is None:
|
|
raise ValidationError("该人生尚未发布")
|
|
active = StoryRun.objects.filter(
|
|
user=user, story_version=version, status=StoryRun.Status.ACTIVE
|
|
).first()
|
|
if active:
|
|
return _node_payload(active)
|
|
run = StoryRun.objects.create(
|
|
user=user,
|
|
story_version=version,
|
|
current_node=version.content["start_node"],
|
|
state=(
|
|
_refresh_mark_counts({"marks": [], "easter_eggs": []}, version.content)
|
|
if version.content.get("system_version") == "marks-v1"
|
|
else {}
|
|
),
|
|
)
|
|
return _node_payload(run)
|
|
|
|
|
|
def get_run_payload(run):
|
|
return _node_payload(run)
|
|
|
|
|
|
@transaction.atomic
|
|
def make_choice(*, run_id, user, choice_index, idempotency_key=None):
|
|
run = (
|
|
StoryRun.objects.select_for_update()
|
|
.select_related("story_version__story")
|
|
.get(id=run_id, user=user)
|
|
)
|
|
if idempotency_key:
|
|
existing = StoryChoice.objects.filter(
|
|
run=run, idempotency_key=idempotency_key
|
|
).first()
|
|
if existing:
|
|
return _node_payload(run)
|
|
if run.status != StoryRun.Status.ACTIVE:
|
|
raise ValidationError("该人生已经结束")
|
|
|
|
node_id = run.current_node
|
|
content = run.story_version.content
|
|
node = content["nodes"][node_id]
|
|
choices = _available_choices(run, node)
|
|
try:
|
|
normalized_index = int(choice_index)
|
|
if normalized_index < 0 or normalized_index >= len(choices):
|
|
raise IndexError
|
|
choice = choices[normalized_index]
|
|
except (IndexError, TypeError, ValueError) as exc:
|
|
raise ValidationError({"choice_index": "选项不存在"}) from exc
|
|
|
|
sequence = run.choices.count() + 1
|
|
effects = choice.get("effects", {})
|
|
StoryChoice.objects.create(
|
|
run=run,
|
|
sequence=sequence,
|
|
idempotency_key=idempotency_key,
|
|
node_id=node_id,
|
|
choice_index=normalized_index,
|
|
choice_text=choice.get("text", ""),
|
|
effects=effects,
|
|
)
|
|
run.state = _merge_effects(run.state, effects)
|
|
if content.get("system_version") == "marks-v1":
|
|
run.state = _refresh_mark_counts(run.state, content)
|
|
mark_lookup = _mark_lookup(content)
|
|
for mark_code in effects.get("marks", []):
|
|
mark = mark_lookup.get(mark_code)
|
|
if mark:
|
|
StoryMark.objects.get_or_create(
|
|
user=user,
|
|
story=run.story_version.story,
|
|
code=mark_code,
|
|
defaults={
|
|
"source_run": run,
|
|
"name": mark["name"],
|
|
"domain": mark["domain"],
|
|
},
|
|
)
|
|
run.current_node = _resolve_ending(content, run.state, choice["next"])
|
|
next_node = content["nodes"][run.current_node]
|
|
if not next_node.get("choices"):
|
|
run.status = StoryRun.Status.COMPLETED
|
|
run.ending_code = run.current_node
|
|
run.completed_at = timezone.now()
|
|
run.save(
|
|
update_fields=[
|
|
"state",
|
|
"current_node",
|
|
"status",
|
|
"ending_code",
|
|
"completed_at",
|
|
"updated_at",
|
|
]
|
|
)
|
|
return _node_payload(run)
|