import re from mission.models import MissionPayloadProcessorEnum STRUCTURED_DESCRIPTION_IMAGE_PREFIXES = ("款式图:", "款式图:") def _extract_labeled_image_url(lines: list[str]) -> tuple[list[str], str]: kept_lines = [] image_url = "" for line in lines: stripped_line = line.strip() matched_prefix = None for prefix in STRUCTURED_DESCRIPTION_IMAGE_PREFIXES: if stripped_line.startswith(prefix): matched_prefix = prefix break if matched_prefix is None: kept_lines.append(line) continue candidate = stripped_line.removeprefix(matched_prefix).strip() if candidate: image_url = candidate 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