feat: message-api for mission via wecomm agent

mission/payload_processors.py

@@ -0,0 +1,86 @@
+import re
+
+from mission.models import MissionPayloadProcessorEnum
+
+
+STRUCTURED_DESCRIPTION_IMAGE_PREFIX = "款式图:"
+
+
+def _extract_labeled_image_url(lines: list[str]) -> tuple[list[str], str]:
+    kept_lines = []
+    image_url = ""
+    for line in lines:
+        stripped_line = line.strip()
+        if stripped_line.startswith(STRUCTURED_DESCRIPTION_IMAGE_PREFIX):
+            candidate = stripped_line.removeprefix(STRUCTURED_DESCRIPTION_IMAGE_PREFIX).strip()
+            if candidate:
+                image_url = candidate
+            continue
+        kept_lines.append(line)
+    return kept_lines, image_url
+
+
+def _split_structured_description(description: str) -> tuple[str, str, str, str]:
+    lines = [line.strip() for line in (description or "").splitlines()]
+    if not lines:
+        return "", "", "", ""
+
+    body_lines = lines[1:] if len(lines) > 1 else []
+    while body_lines and not body_lines[0]:
+        body_lines.pop(0)
+
+    body_lines, image_url = _extract_labeled_image_url(body_lines)
+
+    url = ""
+    if body_lines:
+        last_line = body_lines[-1]
+        match = re.search(r"https?://\S+", last_line)
+        if match is not None:
+            url = match.group(0)
+            body_lines = body_lines[:-1]
+
+    while body_lines and not body_lines[-1]:
+        body_lines.pop()
+
+    split_index = next((index for index, line in enumerate(body_lines) if not line), None)
+    if split_index is None:
+        title_lines = body_lines
+        description_lines = []
+    else:
+        title_lines = body_lines[:split_index]
+        description_lines = body_lines[split_index + 1 :]
+        while description_lines and not description_lines[0]:
+            description_lines.pop(0)
+
+    title = "\n".join(title_lines).strip()
+    parsed_description = "\n".join(description_lines).strip()
+    return title, parsed_description, url, image_url
+
+
+def enhance_structured_description_v1(*, mission, payload: dict, event_key: str) -> dict:
+    title, parsed_description, url, image_url = _split_structured_description(mission.description)
+    return {
+        **payload,
+        "parsed_description_title": title,
+        "parsed_description_body": parsed_description,
+        "parsed_description_url": url,
+        "parsed_description_image_url": image_url,
+    }
+
+
+PAYLOAD_PROCESSORS = {
+    MissionPayloadProcessorEnum.STRUCTURED_DESCRIPTION_V1: enhance_structured_description_v1,
+}
+
+
+def apply_mission_payload_processor(*, mission, payload: dict, event_key: str) -> dict:
+    processor_key = getattr(mission.category, "payload_processor", "") or ""
+    if not processor_key:
+        return payload
+
+    processor = PAYLOAD_PROCESSORS.get(processor_key)
+    if processor is None:
+        return payload
+
+    enhanced_payload = processor(mission=mission, payload=payload, event_key=event_key)
+    return enhanced_payload if isinstance(enhanced_payload, dict) else payload